@Test public void finishingIgnoresExceptions() throws Exception { final StreamTaskTestHarness<String> testHarness = new StreamTaskTestHarness<>( SourceStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO); final CompletableFuture<Void> operatorRunningWaitingFuture = new CompletableFuture<>(); ExceptionThrowingSource.setIsInRunLoopFuture(operatorRunningWaitingFuture); testHarness.setupOutputForSingletonOperatorChain(); StreamConfig streamConfig = testHarness.getStreamConfig(); streamConfig.setStreamOperator(new StreamSource<>(new ExceptionThrowingSource())); streamConfig.setOperatorID(new OperatorID()); testHarness.invoke(); operatorRunningWaitingFuture.get(); testHarness.getTask().finishTask(); testHarness.waitForTaskCompletion(); }
@Override protected void finishTask() throws Exception { isFinished = true; cancelTask(); }
SourceStreamTask extends StreamTask<OUT, OP> { @Override protected void finishTask() throws Exception { isFinished = true; cancelTask(); } }
SourceStreamTask extends StreamTask<OUT, OP> { @Override protected void finishTask() throws Exception { isFinished = true; cancelTask(); } SourceStreamTask(Environment env); }
SourceStreamTask extends StreamTask<OUT, OP> { @Override protected void finishTask() throws Exception { isFinished = true; cancelTask(); } SourceStreamTask(Environment env); @Override Optional<Thread> getExecutingThread(); @Override boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime); }
SourceStreamTask extends StreamTask<OUT, OP> { @Override protected void finishTask() throws Exception { isFinished = true; cancelTask(); } SourceStreamTask(Environment env); @Override Optional<Thread> getExecutingThread(); @Override boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime);  }