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);  }
@Test() public void testInvalidInputData() throws Exception { try { final TaskExecutor taskExecutor = getTaskExecutor(null, null); taskExecutor.execute(); fail(); } catch (NullPointerException e) { assertTrue(true); } }
@Test() public void testUnboundedSourceVertexDataFetching() throws Exception { final IRVertex sourceIRVertex = new TestUnboundedSourceVertex(); final Long watermark = 1234567L; final BlockingQueue<Long> watermarkQueue = new LinkedBlockingQueue<>(); watermarkQueue.add(watermark); final Readable readable = new TestUnboundedSourceReadable(watermarkQueue, 1); final Map<String, Readable> vertexIdToReadable = new HashMap<>(); vertexIdToReadable.put(sourceIRVertex.getId(), readable); final List<Watermark> emittedWatermarks = new LinkedList<>(); final Transform transform = new StreamTransformNoWatermarkEmit(emittedWatermarks); final OperatorVertex operatorVertex = new OperatorVertex(transform); final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(sourceIRVertex) .addVertex(operatorVertex) .connectVertices(createEdge(sourceIRVertex, operatorVertex, "edge1")) .buildWithoutSourceSinkCheck(); final StageEdge taskOutEdge = mockStageEdgeFrom(operatorVertex); final Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.emptyList(), Collections.singletonList(taskOutEdge), vertexIdToReadable); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); assertEquals(Arrays.asList(new Watermark(watermark)), emittedWatermarks); assertEquals(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId())); }
@Test(timeout = 5000) public void testParentTaskDataFetching() throws Exception { final IRVertex vertex = new OperatorVertex(new StreamTransform()); final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(vertex) .buildWithoutSourceSinkCheck(); final StageEdge taskOutEdge = mockStageEdgeFrom(vertex); final Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.singletonList(mockStageEdgeTo(vertex)), Collections.singletonList(taskOutEdge), Collections.emptyMap()); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); assertTrue(checkEqualElements(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId()))); }
@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()))); }
@Test(timeout = 5000) public void testTwoOperators() throws Exception { final IRVertex operatorIRVertex1 = new OperatorVertex(new StreamTransform()); final IRVertex operatorIRVertex2 = new OperatorVertex(new StreamTransform()); 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 Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.singletonList(mockStageEdgeTo(operatorIRVertex1)), Collections.singletonList(taskOutEdge), Collections.emptyMap()); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); assertTrue(checkEqualElements(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId()))); }
@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))); }
@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)); }
@Test() public void testSourceVertexDataFetching() throws Exception { final IRVertex sourceIRVertex = new InMemorySourceVertex<>(elements); final Readable readable = new BoundedIteratorReadable() { @Override protected Iterator initializeIterator() { return elements.iterator(); } @Override public long readWatermark() { throw new UnsupportedOperationException(); } @Override public List<String> getLocations() { throw new UnsupportedOperationException(); } @Override public void close() throws IOException { } }; final Map<String, Readable> vertexIdToReadable = new HashMap<>(); vertexIdToReadable.put(sourceIRVertex.getId(), readable); final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>() .addVertex(sourceIRVertex) .buildWithoutSourceSinkCheck(); final StageEdge taskOutEdge = mockStageEdgeFrom(sourceIRVertex); final Task task = new Task( "testSourceVertexDataFetching", generateTaskId(), TASK_EXECUTION_PROPERTY_MAP, new byte[0], Collections.emptyList(), Collections.singletonList(taskOutEdge), vertexIdToReadable); final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag); taskExecutor.execute(); assertTrue(checkEqualElements(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId()))); }