class ZooKeeperStateHandleStore {

	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();
				}
			}
		}
	}

	public  ZooKeeperStateHandleStore(
		CuratorFramework client,
		RetrievableStateStorageHelper<T> storage);

}