@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 testAddAndLock() throws Exception {
        final TestingLongStateHandleHelper longStateStorage = new TestingLongStateHandleHelper();
        ZooKeeperStateHandleStore<TestingLongStateHandleHelper.LongStateHandle> store =
                new ZooKeeperStateHandleStore<>(getZooKeeperClient(), longStateStorage);

        // Config
        final String pathInZooKeeper = "/testAdd";
        final long state = 1239712317L;

        // Test
        store.addAndLock(pathInZooKeeper, new TestingLongStateHandleHelper.LongStateHandle(state));

        // Verify
        // State handle created
        assertThat(store.getAllAndLock()).hasSize(1);
        assertThat(store.getAndLock(pathInZooKeeper).retrieveState().getValue()).isEqualTo(state);

        // Path created and is persistent
        Stat stat = getZooKeeperClient().checkExists().forPath(pathInZooKeeper);
        assertThat(stat).isNotNull();
        assertThat(stat.getEphemeralOwner()).isZero();

        List<String> children = getZooKeeperClient().getChildren().forPath(pathInZooKeeper);

        // There should be one child which is the locks subfolder
        final String locksSubfolderChild = Iterables.getOnlyElement(children);
        stat =
                getZooKeeperClient()
                        .checkExists()
                        .forPath(generateZookeeperPath(pathInZooKeeper, locksSubfolderChild));
        assertThat(stat).isNotNull();

        assertThat(stat.getEphemeralOwner())
                .as("The lock subfolder shouldn't be ephemeral")
                .isZero();

        List<String> lockChildren =
                getZooKeeperClient()
                        .getChildren()
                        .forPath(generateZookeeperPath(pathInZooKeeper, locksSubfolderChild));
        // Only one lock is expected
        final String lockChild = Iterables.getOnlyElement(lockChildren);
        stat =
                getZooKeeperClient()
                        .checkExists()
                        .forPath(
                                generateZookeeperPath(
                                        pathInZooKeeper, locksSubfolderChild, lockChild));

        assertThat(stat.getEphemeralOwner()).as("The lock node should be ephemeral").isNotZero();

        // Data is equal
        @SuppressWarnings("unchecked")
        final long actual =
                ((RetrievableStateHandle<TestingLongStateHandleHelper.LongStateHandle>)
                                InstantiationUtil.deserializeObject(
                                        getZooKeeperClient().getData().forPath(pathInZooKeeper),
                                        ClassLoader.getSystemClassLoader()))
                        .retrieveState()
                        .getValue();

        assertThat(actual).isEqualTo(state);
    }