@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())); }
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);  }