@Test() public void testMultipleIncomingEdges() throws Exception { final List<Watermark> emittedWatermarks = new ArrayList<>(); final IRVertex operatorIRVertex1 = new OperatorVertex(new StreamTransform()); final IRVertex operatorIRVertex2 = new OperatorVertex(new StreamTransformNoWatermarkEmit(emittedWatermarks)); final IRVertex operatorIRVertex3 = new OperatorVertex(new StreamTransform()); final IRVertex sourceIRVertex1 = new TestUnboundedSourceVertex(); final IRVertex sourceIRVertex2 = new TestUnboundedSourceVertex(); final Queue<Long> watermarks1 = new ConcurrentLinkedQueue<>(); watermarks1.add(500L); final Queue<Long> watermarks2 = new ConcurrentLinkedQueue<>(); watermarks2.add(1000L); final Readable readable1 = new TestUnboundedSourceReadable(watermarks1, 5); final Readable readable2 = new TestUnboundedSourceReadable(watermarks2, 2); final Map<String, Readable> vertexIdToReadable = new HashMap<>(); vertexIdToReadable.put(sourceIRVertex1.getId(), readable1); vertexIdToReadable.put(sourceIRVertex2.getId(), readable2); final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(sourceIRVertex1) .addVertex(sourceIRVertex2) .addVertex(operatorIRVertex1) .addVertex(operatorIRVertex2) .addVertex(operatorIRVertex3) .connectVertices(createEdge(sourceIRVertex1, operatorIRVertex1, "edge1")) .connectVertices(createEdge(operatorIRVertex1, operatorIRVertex2, "edge2")) .connectVertices(createEdge(sourceIRVertex2, operatorIRVertex3, "edge3")) .connectVertices(createEdge(operatorIRVertex3, operatorIRVertex2, "edge4")) .buildWithoutSourceSinkCheck(); final StageEdge taskOutEdge = mockStageEdgeFrom(operatorIRVertex2); final Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.emptyList(), Collections.singletonList(taskOutEdge), vertexIdToReadable); final Thread watermarkEmitThread = new Thread(new Runnable() { @Override public void run() { waitUntilWatermarkEmitted(watermarks2); watermarks1.add(600L); watermarks1.add(1400L); watermarks1.add(1800L); waitUntilWatermarkEmitted(watermarks1); watermarks2.add(2200L); waitUntilWatermarkEmitted(watermarks2); watermarks1.add(2500L); waitUntilWatermarkEmitted(watermarks1); } }); watermarkEmitThread.start(); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); watermarkEmitThread.join(); assertEquals(Arrays.asList( new Watermark(500), new Watermark(600), new Watermark(1000), new Watermark(1800), new Watermark(2200)), emittedWatermarks); final List<Integer> doubledElements = new ArrayList<>(elements.size() * 2); doubledElements.addAll(elements); doubledElements.addAll(elements); assertTrue(checkEqualElements(doubledElements, runtimeEdgeToOutputData.get(taskOutEdge.getId()))); }
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);  }