@Override
    public CompletableFuture<Void> shutdown(JobStatus jobStatus) {
        if (!running) {
            return FutureUtils.completedVoidFuture();
        }
        running = false;

        LOG.info("Shutting down.");
        if (jobStatus.isGloballyTerminalState()) {
            LOG.info("Removing counter from ConfigMap {}", configMapName);
            return kubeClient
                    .checkAndUpdateConfigMap(
                            configMapName,
                            configMap -> {
                                if (isValidOperation(configMap)) {
                                    configMap.getData().remove(CHECKPOINT_COUNTER_KEY);
                                    return Optional.of(configMap);
                                }
                                return Optional.empty();
                            })
                    // checkAndUpdateConfigMap only returns false if the callback returned an empty
                    // ConfigMap. We don't want to continue the cleanup in that case, i.e. we can
                    // ignore the return value
                    .thenApply(valueChanged -> null);
        }

        return FutureUtils.completedVoidFuture();
    }
@Test
    void testShutdown() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderCallbackGrantLeadership();

                            final KubernetesCheckpointIDCounter checkpointIDCounter =
                                    new KubernetesCheckpointIDCounter(
                                            flinkKubeClient, LEADER_CONFIGMAP_NAME, LOCK_IDENTITY);
                            checkpointIDCounter.start();

                            checkpointIDCounter.setCount(100L);

                            assertThat(getLeaderConfigMap().getData())
                                    .containsEntry(Constants.CHECKPOINT_COUNTER_KEY, "100");

                            checkpointIDCounter.shutdown(JobStatus.FINISHED).join();

                            assertThat(getLeaderConfigMap().getData())
                                    .doesNotContainKey(Constants.CHECKPOINT_COUNTER_KEY);
                        });
            }
        };
    }