public final String getName() {
        return getEnvironment().getTaskInfo().getTaskNameWithSubtasks();
    }
@Test
    void testTaskAvoidHangingAfterSnapshotStateThrownException() throws Exception {
        // given: Configured SourceStreamTask with source which fails on checkpoint.
        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(4712L, 43L));

        FailedSource failedSource = new FailedSource();
        cfg.setStreamOperator(new TestStreamSource<String, FailedSource>(failedSource));
        cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);

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

            // when: Task starts
            task.startTaskThread();

            // wait for the task starts doing the work.
            failedSource.awaitRunning();

            // and: Checkpoint is triggered which should lead to exception in Source.
            task.triggerCheckpointBarrier(
                    42L, 1L, CheckpointOptions.forCheckpointWithDefaultLocation());

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

            // then: The task doesn't hang but finished with FAILED state.
            assertThat(task.getExecutionState()).isEqualTo(ExecutionState.FAILED);
        }
    }