@Override
    public RetrievableStateHandle<T> addAndLock(String key, T state)
            throws PossibleInconsistentStateException, Exception {
        checkNotNull(key, "Key in ConfigMap.");
        checkNotNull(state, "State.");

        final RetrievableStateHandle<T> storeHandle = storage.store(state);

        final byte[] serializedStoreHandle =
                serializeOrDiscard(new StateHandleWithDeleteMarker<>(storeHandle));

        // initialize flag to serve the failure case
        boolean discardState = true;
        try {
            // a successful operation will result in the state not being discarded
            discardState =
                    !updateConfigMap(
                                    cm -> {
                                        try {
                                            return addEntry(cm, key, serializedStoreHandle);
                                        } catch (Exception e) {
                                            throw new CompletionException(e);
                                        }
                                    })
                            .get();
            return storeHandle;
        } catch (Exception ex) {
            final Optional<PossibleInconsistentStateException> possibleInconsistentStateException =
                    ExceptionUtils.findThrowable(ex, PossibleInconsistentStateException.class);
            if (possibleInconsistentStateException.isPresent()) {
                // it's unclear whether the state handle metadata was written to the ConfigMap -
                // hence, we don't discard the data
                discardState = false;
                throw possibleInconsistentStateException.get();
            }

            throw ExceptionUtils.findThrowable(ex, AlreadyExistException.class)
                    .orElseThrow(() -> ex);
        } finally {
            if (discardState) {
                storeHandle.discardState();
            }
        }
    }
@Test
    void testAddAndLockWithExistingKey() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderCallbackGrantLeadership();

                            final TestingLongStateHandleHelper.LongStateHandle oldState =
                                    addRegularEntry(getLeaderConfigMap(), key, 1337L);

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

                            final TestingLongStateHandleHelper.LongStateHandle newState =
                                    new TestingLongStateHandleHelper.LongStateHandle(12345L);

                            final String msg =
                                    String.format(
                                            "%s already exists in ConfigMap %s",
                                            key, LEADER_CONFIGMAP_NAME);
                            assertThatThrownBy(
                                            () -> store.addAndLock(key, newState),
                                            "Exception should be thrown.")
                                    .satisfies(
                                            anyCauseMatches(
                                                    StateHandleStore.AlreadyExistException.class,
                                                    msg));

                            // Both initial & new handles should be in the storage (we never clean
                            // it for testing).
                            assertThat(TestingLongStateHandleHelper.getGlobalStorageSize())
                                    .isEqualTo(2);
                            // Only the new one (second entry in the store) should have been
                            // discarded.
                            assertThat(oldState.isDiscarded()).isFalse();
                            assertThat(newState.isDiscarded()).isTrue();
                        });
            }
        };
    }