docs: public RetrievableStateHandle<T> addAndLock(
  			String pathInZooKeeper,
  			T state) throws Exception {
  		checkNotNull(pathInZooKeeper, "Path in ZooKeeper");
  		checkNotNull(state, "State");
  
  		final String path = normalizePath(pathInZooKeeper);
  
  		RetrievableStateHandle<T> storeHandle = storage.store(state);
  
  		boolean success = false;
  
  		try {
  			// Serialize the state handle. This writes the state to the backend.
  			byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle);
  
  			// Write state handle (not the actual state) to ZooKeeper. This is expected to be
  			// smaller than the state itself. This level of indirection makes sure that data in
  			// ZooKeeper is small, because ZooKeeper is designed for data in the KB range, but
  			// the state can be larger.
  			// Create the lock node in a transaction with the actual state node. That way we can prevent
  			// race conditions with a concurrent delete operation.
  			client.inTransaction().create().withMode(CreateMode.PERSISTENT).forPath(path, serializedStoreHandle)
  				.and().create().withMode(CreateMode.EPHEMERAL).forPath(getLockPath(path))
  				.and().commit();
  
  			success = true;
  			return storeHandle;
  		}
  		catch (KeeperException.NodeExistsException e) {
  			throw new ConcurrentModificationException("ZooKeeper unexpectedly modified", e);
  		}
  		finally {
  			if (!success) {
  				// Cleanup the state handle if it was not written to ZooKeeper.
  				if (storeHandle != null) {
  					storeHandle.discardState();
  				}
  			}
  		}
  	}