StreamTask extends AbstractInvokable implements StatefulTask, AsyncExceptionHandler { public String getName() { return getEnvironment().getTaskInfo().getTaskNameWithSubtasks(); }  @VisibleForTesting void setProcessingTimeService(ProcessingTimeService timeProvider); @Override final void invoke(); @Override final void cancel(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override void setInitialState(TaskStateSnapshot taskStateHandles); @Override boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions); @Override void triggerCheckpointOnBarrier(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			CheckpointMetrics checkpointMetrics); @Override void abortCheckpointOnBarrier(long checkpointId, Throwable cause); ExecutorService getAsyncOperationsThreadPool(); @Override void notifyCheckpointComplete(long checkpointId); OperatorStateBackend createOperatorStateBackend(
			StreamOperator<?> op, Collection<OperatorStateHandle> restoreStateHandles); AbstractKeyedStateBackend<K> createKeyedStateBackend(
			TypeSerializer<K> keySerializer,
			int numberOfKeyGroups,
			KeyGroupRange keyGroupRange); CheckpointStreamFactory createCheckpointStreamFactory(StreamOperator<?> operator); CheckpointStreamFactory createSavepointStreamFactory(StreamOperator<?> operator, String targetLocation); ProcessingTimeService getProcessingTimeService(); @Override void handleAsyncException(String message, Throwable exception); @Override String toString(); CloseableRegistry getCancelables(); static final ThreadGroup TRIGGER_THREAD_GROUP; }
@Test public void testStateBackendLoadingAndClosing() throws Exception { Configuration taskManagerConfig = new Configuration(); taskManagerConfig.setString(CoreOptions.STATE_BACKEND, MockStateBackend.class.getName()); StreamConfig cfg = new StreamConfig(new Configuration()); cfg.setOperatorID(new OperatorID(4711L, 42L)); cfg.setStreamOperator(new StreamSource<>(new MockSourceFunction())); cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime); Task task = createTask(StateBackendTestSource.class, cfg, taskManagerConfig); StateBackendTestSource.fail = false; task.startTaskThread(); task.getExecutingThread().join(); verify(StateBackendTestSource.operatorStateBackend).close(); verify(StateBackendTestSource.keyedStateBackend).close(); assertEquals(ExecutionState.FINISHED, task.getExecutionState()); }
@Test public void testStateBackendClosingOnFailure() throws Exception { Configuration taskManagerConfig = new Configuration(); taskManagerConfig.setString(CoreOptions.STATE_BACKEND, MockStateBackend.class.getName()); StreamConfig cfg = new StreamConfig(new Configuration()); cfg.setOperatorID(new OperatorID(4711L, 42L)); cfg.setStreamOperator(new StreamSource<>(new MockSourceFunction())); cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime); Task task = createTask(StateBackendTestSource.class, cfg, taskManagerConfig); StateBackendTestSource.fail = true; task.startTaskThread(); task.getExecutingThread().join(); verify(StateBackendTestSource.operatorStateBackend).close(); verify(StateBackendTestSource.keyedStateBackend).close(); assertEquals(ExecutionState.FAILED, task.getExecutionState()); }