public DiscardObject markAsDiscardedOnShutdown(JobStatus jobStatus) {
        return shouldBeDiscardedOnShutdown(jobStatus) ? markAsDiscarded() : NOOP_DISCARD_OBJECT;
    }
@Test
    void testCleanUpOnShutdown() throws Exception {
        JobStatus[] terminalStates =
                new JobStatus[] {
                    JobStatus.FINISHED, JobStatus.CANCELED, JobStatus.FAILED, JobStatus.SUSPENDED
                };

        for (JobStatus status : terminalStates) {

            OperatorState state = mock(OperatorState.class);
            Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
            operatorStates.put(new OperatorID(), state);

            EmptyStreamStateHandle retainedHandle = new EmptyStreamStateHandle();
            TestCompletedCheckpointStorageLocation retainedLocation =
                    new TestCompletedCheckpointStorageLocation(retainedHandle, "ptr");

            // Keep
            CheckpointProperties retainProps =
                    new CheckpointProperties(
                            false,
                            CheckpointType.CHECKPOINT,
                            false,
                            false,
                            false,
                            false,
                            false,
                            false);
            CompletedCheckpoint checkpoint =
                    new CompletedCheckpoint(
                            new JobID(),
                            0,
                            0,
                            1,
                            new HashMap<>(operatorStates),
                            Collections.emptyList(),
                            retainProps,
                            retainedLocation,
                            null);

            checkpoint.markAsDiscardedOnShutdown(status).discard();

            verify(state, times(0)).discardState();
            assertThat(retainedLocation.isDisposed()).isFalse();
            assertThat(retainedHandle.isDisposed()).isFalse();

            // Discard
            EmptyStreamStateHandle discardHandle = new EmptyStreamStateHandle();
            TestCompletedCheckpointStorageLocation discardLocation =
                    new TestCompletedCheckpointStorageLocation(discardHandle, "ptr");

            // Keep
            CheckpointProperties discardProps =
                    new CheckpointProperties(
                            false, CheckpointType.CHECKPOINT, true, true, true, true, true, false);

            checkpoint =
                    new CompletedCheckpoint(
                            new JobID(),
                            0,
                            0,
                            1,
                            new HashMap<>(operatorStates),
                            Collections.emptyList(),
                            discardProps,
                            discardLocation,
                            null);

            checkpoint.markAsDiscardedOnShutdown(status).discard();

            verify(state, times(1)).discardState();
            assertThat(discardLocation.isDisposed()).isTrue();
            assertThat(discardHandle.isDisposed()).isTrue();
        }
    }