### Question:
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { public String getName() { return getEnvironment().getTaskInfo().getTaskNameWithSubtasks(); } protected  StreamTask(Environment env); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorage getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @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); 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; }### Answer:
@Test public void testStateBackendLoadingAndClosing() throws Exception { Configuration taskManagerConfig = new Configuration(); taskManagerConfig.setString(CheckpointingOptions.STATE_BACKEND, TestMemoryStateBackendFactory.class.getName()); StreamConfig cfg = new StreamConfig(new Configuration()); cfg.setStateKeySerializer(mock(TypeSerializer.class)); cfg.setOperatorID(new OperatorID(4711L, 42L)); TestStreamSource<Long, MockSourceFunction> streamSource = new TestStreamSource<>(new MockSourceFunction()); cfg.setStreamOperator(streamSource); cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime); Task task = createTask(StateBackendTestSource.class, cfg, taskManagerConfig); StateBackendTestSource.fail = false; task.startTaskThread(); task.getExecutingThread().join(); verify(TestStreamSource.operatorStateBackend).close(); verify(TestStreamSource.keyedStateBackend).close(); verify(TestStreamSource.rawOperatorStateInputs).close(); verify(TestStreamSource.rawKeyedStateInputs).close(); verify(TestStreamSource.operatorStateBackend).dispose(); verify(TestStreamSource.keyedStateBackend).dispose(); assertEquals(ExecutionState.FINISHED, task.getExecutionState()); }
@Test public void testStateBackendClosingOnFailure() throws Exception { Configuration taskManagerConfig = new Configuration(); taskManagerConfig.setString(CheckpointingOptions.STATE_BACKEND, TestMemoryStateBackendFactory.class.getName()); StreamConfig cfg = new StreamConfig(new Configuration()); cfg.setStateKeySerializer(mock(TypeSerializer.class)); cfg.setOperatorID(new OperatorID(4711L, 42L)); TestStreamSource<Long, MockSourceFunction> streamSource = new TestStreamSource<>(new MockSourceFunction()); cfg.setStreamOperator(streamSource); cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime); Task task = createTask(StateBackendTestSource.class, cfg, taskManagerConfig); StateBackendTestSource.fail = true; task.startTaskThread(); task.getExecutingThread().join(); verify(TestStreamSource.operatorStateBackend).close(); verify(TestStreamSource.keyedStateBackend).close(); verify(TestStreamSource.rawOperatorStateInputs).close(); verify(TestStreamSource.rawKeyedStateInputs).close(); verify(TestStreamSource.operatorStateBackend).dispose(); verify(TestStreamSource.keyedStateBackend).dispose(); assertEquals(ExecutionState.FAILED, task.getExecutionState()); }