public StreamConfig getConfiguration() {
        return configuration;
    }
@Test
    void testEarlyCanceling() throws Exception {
        final StreamConfig cfg = new StreamConfig(new Configuration());
        cfg.setOperatorID(new OperatorID(4711L, 42L));
        cfg.setStreamOperator(new SlowlyDeserializingOperator());
        cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        cfg.serializeAllConfigs();

        final TaskManagerActions taskManagerActions = spy(new NoOpTaskManagerActions());
        try (NettyShuffleEnvironment shuffleEnvironment =
                new NettyShuffleEnvironmentBuilder().build()) {
            final Task task =
                    new TestTaskBuilder(shuffleEnvironment)
                            .setInvokable(SourceStreamTask.class)
                            .setTaskConfig(cfg.getConfiguration())
                            .setTaskManagerActions(taskManagerActions)
                            .build(EXECUTOR_EXTENSION.getExecutor());

            final TaskExecutionState state =
                    new TaskExecutionState(task.getExecutionId(), ExecutionState.RUNNING);

            task.startTaskThread();

            verify(taskManagerActions, timeout(2000L)).updateTaskExecutionState(eq(state));

            // send a cancel. because the operator takes a long time to deserialize, this should
            // hit the task before the operator is deserialized
            task.cancelExecution();

            task.getExecutingThread().join();

            assertThat(task.getExecutingThread().isAlive()).as("Task did not cancel").isFalse();
            assertThat(task.getExecutionState()).isEqualTo(ExecutionState.CANCELED);
        }
    }