@Override
    public StringResourceVersion exists(String key) throws Exception {
        checkNotNull(key, "Key in ConfigMap.");

        return kubeClient
                .getConfigMap(configMapName)
                .map(
                        configMap -> {
                            final String content = configMap.getData().get(key);
                            if (content != null) {
                                try {
                                    final StateHandleWithDeleteMarker<T> stateHandle =
                                            deserializeStateHandle(content);
                                    if (stateHandle.isMarkedForDeletion()) {
                                        return StringResourceVersion.notExisting();
                                    }
                                } catch (IOException e) {
                                    // Any calls to add or replace will try to remove this resource,
                                    // so we can simply treat it as non-existent.
                                    return StringResourceVersion.notExisting();
                                }
                                return StringResourceVersion.valueOf(
                                        configMap.getResourceVersion());
                            }
                            return StringResourceVersion.notExisting();
                        })
                .orElseThrow(this::getConfigMapNotExistException);
    }
@Test
    void testExistsWithDeletingEntry() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderCallbackGrantLeadership();

                            final TestingLongStateHandleHelper.LongStateHandle state =
                                    addDeletingEntry(getLeaderConfigMap(), key, 1337L);

                            final KubernetesStateHandleStore<
                                            TestingLongStateHandleHelper.LongStateHandle>
                                    store =
                                            new KubernetesStateHandleStore<>(
                                                    flinkKubeClient,
                                                    LEADER_CONFIGMAP_NAME,
                                                    longStateStorage,
                                                    filter,
                                                    LOCK_IDENTITY);

                            final StringResourceVersion resourceVersion = store.exists(key);
                            assertThat(resourceVersion.isExisting()).isFalse();
                            // We don't try to pro-actively remove the entry here.
                            assertThat(state.isDiscarded()).isFalse();
                        });
            }
        };
    }