@Override
    public void replace(String pathInZooKeeper, IntegerResourceVersion expectedVersion, T state)
            throws Exception {
        checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
        checkNotNull(state, "State");

        final String path = normalizePath(pathInZooKeeper);

        checkState(
                hasLock(path),
                "'{}' is only allowed to be replaced if the instance has a lock on this node.",
                path);

        RetrievableStateHandle<T> oldStateHandle = get(path, false);

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

        final byte[] serializedStateHandle = serializeOrDiscard(newStateHandle);

        // initialize flags to serve the failure case
        boolean discardOldState = false;
        boolean discardNewState = true;
        try {
            setStateHandle(path, serializedStateHandle, expectedVersion.getValue());

            // swap subject for deletion in case of success
            discardOldState = true;
            discardNewState = false;
        } catch (Exception e) {
            if (indicatesPossiblyInconsistentState(e)) {
                // it's unclear whether the state handle metadata was written to ZooKeeper -
                // hence, we don't discard any data
                discardNewState = false;
                throw new PossibleInconsistentStateException(e);
            }

            // We wrap the exception here so that it could be caught in DefaultJobGraphStore
            throw ExceptionUtils.findThrowable(e, KeeperException.NoNodeException.class)
                    .map(
                            nnee ->
                                    new NotExistException(
                                            "ZooKeeper node " + path + " does not exist.", nnee))
                    .orElseThrow(() -> e);
        } finally {
            if (discardOldState) {
                oldStateHandle.discardState();
            }

            if (discardNewState) {
                newStateHandle.discardState();
            }
        }
    }
@Test
    void testReplace() throws Exception {
        // Setup
        final TestingLongStateHandleHelper stateHandleProvider = new TestingLongStateHandleHelper();

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

        // Config
        final String pathInZooKeeper = "/testReplace";
        final long initialState = 30968470898L;
        final long replaceState = 88383776661L;

        // Test
        store.addAndLock(
                pathInZooKeeper, new TestingLongStateHandleHelper.LongStateHandle(initialState));
        store.replace(
                pathInZooKeeper,
                IntegerResourceVersion.valueOf(0),
                new TestingLongStateHandleHelper.LongStateHandle(replaceState));

        // Verify
        // State handles created
        assertThat(TestingLongStateHandleHelper.getGlobalStorageSize()).isEqualTo(2);
        assertThat(TestingLongStateHandleHelper.getStateHandleValueByIndex(0))
                .isEqualTo(initialState);
        assertThat(TestingLongStateHandleHelper.getStateHandleValueByIndex(1))
                .isEqualTo(replaceState);

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

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

        assertThat(actual).isEqualTo(replaceState);
    }