@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 testReplaceNonExistingPath() {
        final RetrievableStateStorageHelper<TestingLongStateHandleHelper.LongStateHandle>
                stateStorage = new TestingLongStateHandleHelper();

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

        assertThatThrownBy(
                        () ->
                                store.replace(
                                        "/testReplaceNonExistingPath",
                                        IntegerResourceVersion.valueOf(0),
                                        new TestingLongStateHandleHelper.LongStateHandle(1L)))
                .isInstanceOf(Exception.class);
    }