@Nullable
    @Override
    public CompletedCheckpoint addCheckpointAndSubsumeOldestOne(
            CompletedCheckpoint checkpoint,
            CheckpointsCleaner checkpointsCleaner,
            Runnable postCleanup)
            throws Exception {

        checkpoints.addLast(checkpoint);

        CompletedCheckpoint completedCheckpoint =
                CheckpointSubsumeHelper.subsume(
                                checkpoints,
                                maxNumberOfCheckpointsToRetain,
                                (cc) -> {
                                    cc.markAsDiscardedOnSubsume();
                                    checkpointsCleaner.addSubsumedCheckpoint(cc);
                                })
                        .orElse(null);

        findLowest(checkpoints)
                .ifPresent(
                        id ->
                                checkpointsCleaner.cleanSubsumedCheckpoints(
                                        id,
                                        getSharedStateRegistry().unregisterUnusedState(id),
                                        postCleanup,
                                        ioExecutor));

        return completedCheckpoint;
    }
@Test
    void testAddCheckpointWithFailedRemove() throws Exception {

        final int numCheckpointsToRetain = 1;
        CompletedCheckpointStore store =
                createRecoveredCompletedCheckpointStore(
                        numCheckpointsToRetain, Executors.directExecutor());

        CountDownLatch discardAttempted = new CountDownLatch(1);
        for (long i = 0; i < numCheckpointsToRetain + 1; ++i) {
            CompletedCheckpoint checkpointToAdd =
                    new CompletedCheckpoint(
                            new JobID(),
                            i,
                            i,
                            i,
                            Collections.emptyMap(),
                            Collections.emptyList(),
                            CheckpointProperties.forCheckpoint(NEVER_RETAIN_AFTER_TERMINATION),
                            new TestCompletedCheckpointStorageLocation(),
                            null) {
                        @Override
                        public CompletedCheckpointDiscardObject markAsDiscardedOnSubsume() {
                            discardAttempted.countDown();
                            throw new RuntimeException();
                        }
                    };
            // should fail despite the exception
            store.addCheckpointAndSubsumeOldestOne(
                    checkpointToAdd, new CheckpointsCleaner(), () -> {});
        }
        discardAttempted.await();
    }