@Test public void testUncaughtExceptionInAsynchronousCheckpointingOperation() throws Exception { final RuntimeException failingCause = new RuntimeException("Test exception"); FailingDummyEnvironment failingDummyEnvironment = new FailingDummyEnvironment(failingCause); OperatorSnapshotFutures operatorSnapshotResult = new OperatorSnapshotFutures( ExceptionallyDoneFuture.of(failingCause), DoneFuture.of(SnapshotResult.empty()), DoneFuture.of(SnapshotResult.empty()), DoneFuture.of(SnapshotResult.empty())); final TestingUncaughtExceptionHandler uncaughtExceptionHandler = new TestingUncaughtExceptionHandler(); RunningTask<MockStreamTask> task = runTask(() -> new MockStreamTask( failingDummyEnvironment, operatorChain(streamOperatorWithSnapshot(operatorSnapshotResult)), uncaughtExceptionHandler)); MockStreamTask streamTask = task.streamTask; waitTaskIsRunning(streamTask, task.invocationFuture); streamTask.triggerCheckpoint( new CheckpointMetaData(42L, 1L), CheckpointOptions.forCheckpointWithDefaultLocation(), false); final Throwable uncaughtException = uncaughtExceptionHandler.waitForUncaughtException(); assertThat(uncaughtException, is(failingCause)); streamTask.finishInput(); task.waitForTaskCompletion(false); }
@Override public boolean triggerCheckpoint( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics() .setBytesBufferedInAlignment(0L) .setAlignmentDurationNanos(0L); return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime); } catch (Exception e) { if (isRunning) { throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " + getName() + '.', e); } else { LOG.debug("Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e); return false; } } }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public boolean triggerCheckpoint( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics() .setBytesBufferedInAlignment(0L) .setAlignmentDurationNanos(0L); return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime); } catch (Exception e) { if (isRunning) { throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " + getName() + '.', e); } else { LOG.debug("Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e); return false; } } } }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public boolean triggerCheckpoint( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics() .setBytesBufferedInAlignment(0L) .setAlignmentDurationNanos(0L); return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime); } catch (Exception e) { if (isRunning) { throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " + getName() + '.', e); } else { LOG.debug("Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e); return false; } } } protected  StreamTask(Environment env); protected  StreamTask(Environment env, @Nullable ProcessingTimeService timeProvider); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider,
			Thread.UncaughtExceptionHandler uncaughtExceptionHandler); }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public boolean triggerCheckpoint( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics() .setBytesBufferedInAlignment(0L) .setAlignmentDurationNanos(0L); return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime); } catch (Exception e) { if (isRunning) { throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " + getName() + '.', e); } else { LOG.debug("Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e); return false; } } } protected  StreamTask(Environment env); protected  StreamTask(Environment env, @Nullable ProcessingTimeService timeProvider); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider,
			Thread.UncaughtExceptionHandler uncaughtExceptionHandler); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); MailboxExecutor getTaskMailboxExecutor(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorageWorkerView getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override boolean triggerCheckpoint(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override void triggerCheckpointOnBarrier(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			CheckpointMetrics checkpointMetrics); @Override void abortCheckpointOnBarrier(long checkpointId, Throwable cause); ExecutorService getAsyncOperationsThreadPool(); @Override void notifyCheckpointComplete(long checkpointId); ProcessingTimeService getProcessingTimeService(); @Override void handleAsyncException(String message, Throwable exception); @Override String toString(); CloseableRegistry getCancelables(); @VisibleForTesting static List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(
			StreamConfig configuration,
			Environment environment); }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public boolean triggerCheckpoint( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) throws Exception { try { CheckpointMetrics checkpointMetrics = new CheckpointMetrics() .setBytesBufferedInAlignment(0L) .setAlignmentDurationNanos(0L); return performCheckpoint(checkpointMetaData, checkpointOptions, checkpointMetrics, advanceToEndOfEventTime); } catch (Exception e) { if (isRunning) { throw new Exception("Could not perform checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " + getName() + '.', e); } else { LOG.debug("Could not perform checkpoint {} for operator {} while the " + "invokable was not in state running.", checkpointMetaData.getCheckpointId(), getName(), e); return false; } } } protected  StreamTask(Environment env); protected  StreamTask(Environment env, @Nullable ProcessingTimeService timeProvider); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider,
			Thread.UncaughtExceptionHandler uncaughtExceptionHandler); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); MailboxExecutor getTaskMailboxExecutor(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorageWorkerView getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override boolean triggerCheckpoint(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @Override void triggerCheckpointOnBarrier(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			CheckpointMetrics checkpointMetrics); @Override void abortCheckpointOnBarrier(long checkpointId, Throwable cause); ExecutorService getAsyncOperationsThreadPool(); @Override void notifyCheckpointComplete(long checkpointId); ProcessingTimeService getProcessingTimeService(); @Override void handleAsyncException(String message, Throwable exception); @Override String toString(); CloseableRegistry getCancelables(); @VisibleForTesting static List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(
			StreamConfig configuration,
			Environment environment); static final ThreadGroup TRIGGER_THREAD_GROUP; }