@Override
    public long getAndIncrement() throws Exception {
        final AtomicLong current = new AtomicLong();
        final boolean updated =
                kubeClient
                        .checkAndUpdateConfigMap(
                                configMapName,
                                configMap -> {
                                    if (isValidOperation(configMap)) {
                                        final long currentValue = getCurrentCounter(configMap);
                                        current.set(currentValue);
                                        configMap
                                                .getData()
                                                .put(
                                                        CHECKPOINT_COUNTER_KEY,
                                                        String.valueOf(currentValue + 1));
                                        return Optional.of(configMap);
                                    }
                                    return Optional.empty();
                                })
                        .get();

        if (updated) {
            return current.get();
        } else {
            throw new KubernetesException(
                    "Failed to update ConfigMap "
                            + configMapName
                            + " since current KubernetesCheckpointIDCounter does not have the leadership.");
        }
    }
@Test
    void testGetAndIncrement() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderCallbackGrantLeadership();

                            final KubernetesCheckpointIDCounter checkpointIDCounter =
                                    new KubernetesCheckpointIDCounter(
                                            flinkKubeClient, LEADER_CONFIGMAP_NAME, LOCK_IDENTITY);
                            checkpointIDCounter.setCount(100L);
                            final long counter = checkpointIDCounter.getAndIncrement();
                            assertThat(counter).isEqualTo(100L);
                            assertThat(checkpointIDCounter.get()).isEqualTo(101L);
                        });
            }
        };
    }