@Override
    public RetrievableStateHandle<T> addAndLock(String pathInZooKeeper, T state)
            throws PossibleInconsistentStateException, Exception {
        checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
        checkNotNull(state, "State");
        final String path = normalizePath(pathInZooKeeper);
        final Optional<Stat> maybeStat = getStat(path);

        if (maybeStat.isPresent()) {
            if (isNotMarkedForDeletion(maybeStat.get())) {
                throw new AlreadyExistException(
                        String.format("ZooKeeper node %s already exists.", path));
            }

            Preconditions.checkState(
                    releaseAndTryRemove(path),
                    "The state is marked for deletion and, therefore, should be deletable.");
        }

        final RetrievableStateHandle<T> storeHandle = storage.store(state);
        final byte[] serializedStoreHandle = serializeOrDiscard(storeHandle);
        try {
            writeStoreHandleTransactionally(path, serializedStoreHandle);
            return storeHandle;
        } catch (KeeperException.NodeExistsException e) {
            // Transactions are not idempotent in the curator version we're currently using, so it
            // is actually possible that we've re-tried a transaction that has already succeeded.
            // We've ensured that the node hasn't been present prior executing the transaction, so
            // we can assume that this is a result of the retry mechanism.
            return storeHandle;
        } catch (Exception e) {
            if (indicatesPossiblyInconsistentState(e)) {
                throw new PossibleInconsistentStateException(e);
            }
            // In case of any other failure, discard the state and rethrow the exception.
            storeHandle.discardState();
            throw e;
        }
    }
@Test
    void testFailingAddWithPossiblyInconsistentState() {
        final TestingLongStateHandleHelper stateHandleProvider = new TestingLongStateHandleHelper();

        CuratorFramework client = spy(getZooKeeperClient());
        when(client.inTransaction()).thenThrow(new RuntimeException("Expected test Exception."));

        ZooKeeperStateHandleStore<TestingLongStateHandleHelper.LongStateHandle> store =
                new ZooKeeperStateHandleStore<>(client, stateHandleProvider);

        // Config
        final String pathInZooKeeper = "/testAddDiscardStateHandleAfterFailure";
        final long state = 81282227L;

        assertThatExceptionOfType(PossibleInconsistentStateException.class)
                .as("PossibleInconsistentStateException should have been thrown.")
                .isThrownBy(
                        () ->
                                store.addAndLock(
                                        pathInZooKeeper,
                                        new TestingLongStateHandleHelper.LongStateHandle(state)));

        // State handle created and not discarded
        assertThat(TestingLongStateHandleHelper.getGlobalStorageSize()).isOne();
        assertThat(TestingLongStateHandleHelper.getStateHandleValueByIndex(0)).isEqualTo(state);
        assertThat(TestingLongStateHandleHelper.getDiscardCallCountForStateHandleByIndex(0))
                .isZero();
    }