@Test(timeout = 5000) public void testAdditionalOutputs() throws Exception { final String additionalTag1 = "bonus1"; final String additionalTag2 = "bonus2"; final IRVertex routerVertex = new OperatorVertex( new RoutingTransform(Arrays.asList(additionalTag1, additionalTag2))); final IRVertex mainVertex = new OperatorVertex(new StreamTransform()); final IRVertex bonusVertex1 = new OperatorVertex(new StreamTransform()); final IRVertex bonusVertex2 = new OperatorVertex(new StreamTransform()); final RuntimeEdge<IRVertex> edge1 = createEdge(routerVertex, mainVertex, "edge-1"); final RuntimeEdge<IRVertex> edge2 = createEdge(routerVertex, bonusVertex1, "edge-2"); final RuntimeEdge<IRVertex> edge3 = createEdge(routerVertex, bonusVertex2, "edge-3"); edge2.getExecutionProperties().put(AdditionalOutputTagProperty.of(additionalTag1)); edge3.getExecutionProperties().put(AdditionalOutputTagProperty.of(additionalTag2)); final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(routerVertex) .addVertex(mainVertex) .addVertex(bonusVertex1) .addVertex(bonusVertex2) .connectVertices(edge1) .connectVertices(edge2) .connectVertices(edge3) .buildWithoutSourceSinkCheck(); final StageEdge outEdge1 = mockStageEdgeFrom(mainVertex); final StageEdge outEdge2 = mockStageEdgeFrom(bonusVertex1); final StageEdge outEdge3 = mockStageEdgeFrom(bonusVertex2); final Task task = new Task( "testAdditionalOutputs", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.singletonList(mockStageEdgeTo(routerVertex)), Arrays.asList(outEdge1, outEdge2, outEdge3), Collections.emptyMap()); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); final List<Integer> mainOutputs = runtimeEdgeToOutputData.get(outEdge1.getId()); final List<Integer> bonusOutputs1 = runtimeEdgeToOutputData.get(outEdge2.getId()); final List<Integer> bonusOutputs2 = runtimeEdgeToOutputData.get(outEdge3.getId()); List<Integer> even = elements.stream().filter(i -> i % 2 == 0).collect(Collectors.toList()); List<Integer> odd = elements.stream().filter(i -> i % 2 != 0).collect(Collectors.toList()); assertTrue(checkEqualElements(even, mainOutputs)); assertTrue(checkEqualElements(odd, bonusOutputs1)); assertTrue(checkEqualElements(odd, bonusOutputs2)); }
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);  }