@Test(timeout = 5000) public void testTwoOperatorsWithBroadcastVariable() { final Transform singleListTransform = new CreateSingleListTransform(); final long broadcastId = 0; final IRVertex operatorIRVertex1 = new OperatorVertex(new StreamTransform()); final IRVertex operatorIRVertex2 = new OperatorVertex(new BroadcastVariablePairingTransform(broadcastId)); final String edgeId = "edge"; final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(operatorIRVertex1) .addVertex(operatorIRVertex2) .connectVertices(createEdge(operatorIRVertex1, operatorIRVertex2, edgeId)) .buildWithoutSourceSinkCheck(); final StageEdge taskOutEdge = mockStageEdgeFrom(operatorIRVertex2); final StageEdge taskInEdge = mockStageEdgeTo(operatorIRVertex1); when(broadcastManagerWorker.get(broadcastId)).thenReturn(new ArrayList<>(elements)); final Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.singletonList(taskInEdge), Collections.singletonList(taskOutEdge), Collections.emptyMap()); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); final List<Pair<List<Integer>, Integer>> pairs = runtimeEdgeToOutputData.get(taskOutEdge.getId()); final List<Integer> values = pairs.stream().map(Pair::right).collect(Collectors.toList()); assertTrue(checkEqualElements(elements, values)); assertTrue(pairs.stream().map(Pair::left).allMatch(broadcastVar -> checkEqualElements(broadcastVar, values))); }
public void execute() { try { doExecute(); } catch (Throwable throwable) { taskStateManager.onTaskStateChanged(TaskState.State.FAILED, Optional.empty(), Optional.empty()); LOG.error(ExceptionUtils.getStackTrace(throwable)); } }
TaskExecutor { public void execute() { try { doExecute(); } catch (Throwable throwable) { taskStateManager.onTaskStateChanged(TaskState.State.FAILED, Optional.empty(), Optional.empty()); LOG.error(ExceptionUtils.getStackTrace(throwable)); } } }
TaskExecutor { public void execute() { try { doExecute(); } catch (Throwable throwable) { taskStateManager.onTaskStateChanged(TaskState.State.FAILED, Optional.empty(), Optional.empty()); LOG.error(ExceptionUtils.getStackTrace(throwable)); } } TaskExecutor(final Task task,
                      final DAG<IRVertex, RuntimeEdge<IRVertex>> irVertexDag,
                      final TaskStateManager taskStateManager,
                      final IntermediateDataIOFactory intermediateDataIOFactory,
                      final BroadcastManagerWorker broadcastManagerWorker,
                      final MetricMessageSender metricMessageSender,
                      final PersistentConnectionToMasterMap persistentConnectionToMasterMap); }
TaskExecutor { public void execute() { try { doExecute(); } catch (Throwable throwable) { taskStateManager.onTaskStateChanged(TaskState.State.FAILED, Optional.empty(), Optional.empty()); LOG.error(ExceptionUtils.getStackTrace(throwable)); } } TaskExecutor(final Task task,
                      final DAG<IRVertex, RuntimeEdge<IRVertex>> irVertexDag,
                      final TaskStateManager taskStateManager,
                      final IntermediateDataIOFactory intermediateDataIOFactory,
                      final BroadcastManagerWorker broadcastManagerWorker,
                      final MetricMessageSender metricMessageSender,
                      final PersistentConnectionToMasterMap persistentConnectionToMasterMap); void execute(); void setIRVertexPutOnHold(final IRVertex irVertex); }
TaskExecutor { public void execute() { try { doExecute(); } catch (Throwable throwable) { taskStateManager.onTaskStateChanged(TaskState.State.FAILED, Optional.empty(), Optional.empty()); LOG.error(ExceptionUtils.getStackTrace(throwable)); } } TaskExecutor(final Task task,
                      final DAG<IRVertex, RuntimeEdge<IRVertex>> irVertexDag,
                      final TaskStateManager taskStateManager,
                      final IntermediateDataIOFactory intermediateDataIOFactory,
                      final BroadcastManagerWorker broadcastManagerWorker,
                      final MetricMessageSender metricMessageSender,
                      final PersistentConnectionToMasterMap persistentConnectionToMasterMap); void execute(); void setIRVertexPutOnHold(final IRVertex irVertex);  }