public final String getName() {
        return getEnvironment().getTaskInfo().getTaskNameWithSubtasks();
    }
@Test
    void testStateBackendClosingOnFailure() throws Exception {
        Configuration taskManagerConfig = new Configuration();
        taskManagerConfig.set(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);

        try (NettyShuffleEnvironment shuffleEnvironment =
                new NettyShuffleEnvironmentBuilder().build()) {
            Task task =
                    createTask(
                            StateBackendTestSource.class,
                            shuffleEnvironment,
                            cfg,
                            taskManagerConfig,
                            EXECUTOR_EXTENSION.getExecutor());

            StateBackendTestSource.fail = true;
            task.startTaskThread();

            // wait for clean termination
            task.getExecutingThread().join();

            // ensure that the state backends and stream iterables are closed ...
            verify(TestStreamSource.operatorStateBackend).close();
            verify(TestStreamSource.keyedStateBackend).close();
            verify(TestStreamSource.rawOperatorStateInputs).close();
            verify(TestStreamSource.rawKeyedStateInputs).close();
            // ... and disposed
            verify(TestStreamSource.operatorStateBackend).dispose();
            verify(TestStreamSource.keyedStateBackend).dispose();

            assertThat(task.getExecutionState()).isEqualTo(ExecutionState.FAILED);
        }
    }