@Override
    public Collection<String> getAllHandles() throws Exception {

        return kubeClient
                .getConfigMap(configMapName)
                .map(
                        configMap ->
                                configMap.getData().keySet().stream()
                                        .filter(configMapKeyFilter)
                                        .filter(
                                                k -> {
                                                    try {
                                                        final String content =
                                                                Objects.requireNonNull(
                                                                        configMap.getData().get(k));
                                                        return !deserializeStateHandle(content)
                                                                .isMarkedForDeletion();
                                                    } catch (IOException e) {
                                                        return false;
                                                    }
                                                })
                                        .collect(Collectors.toList()))
                .orElseThrow(this::getConfigMapNotExistException);
    }
@Test
    void testGetAllHandles() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderCallbackGrantLeadership();

                            final KubernetesStateHandleStore<
                                            TestingLongStateHandleHelper.LongStateHandle>
                                    store =
                                            new KubernetesStateHandleStore<>(
                                                    flinkKubeClient,
                                                    LEADER_CONFIGMAP_NAME,
                                                    longStateStorage,
                                                    filter,
                                                    LOCK_IDENTITY);
                            final List<String> expected = Arrays.asList(key + 3, key + 2, key + 1);

                            for (String each : expected) {
                                store.addAndLock(each, state);
                            }
                            final String[] actual = store.getAllHandles().toArray(new String[0]);
                            expected.sort(Comparator.comparing(e -> e));
                            assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
                        });
            }
        };
    }