@Test
public void testAddAndLock() throws Exception {

	LongStateStorage longStateStorage = new LongStateStorage();
	ZooKeeperStateHandleStore<Long> store = new ZooKeeperStateHandleStore<Long>(
			ZOOKEEPER.getClient(), longStateStorage, Executors.directExecutor());
	// Config
	final String pathInZooKeeper = "/testAdd";
	final Long state = 1239712317L;
	// Test
	store.addAndLock(pathInZooKeeper, state);
	// Verify
	// State handle created
	assertEquals(1, store.getAllAndLock().size());
	assertEquals(state, store.getAndLock(pathInZooKeeper).retrieveState());
	// Path created and is persistent
	Stat stat = ZOOKEEPER.getClient().checkExists().forPath(pathInZooKeeper);
	assertNotNull(stat);
	assertEquals(0, stat.getEphemeralOwner());
	List<String> children = ZOOKEEPER.getClient().getChildren().forPath(pathInZooKeeper);
	// there should be one child which is the lock
	assertEquals(1, children.size());
	stat = ZOOKEEPER.getClient().checkExists().forPath(pathInZooKeeper + '/' + children.get(0));
	assertNotNull(stat);
	// check that the child is an ephemeral node
	assertNotEquals(0, stat.getEphemeralOwner());
	// Data is equal
	@SuppressWarnings("unchecked")
	Long actual = ((RetrievableStateHandle<Long>) InstantiationUtil.deserializeObject(
			ZOOKEEPER.getClient().getData().forPath(pathInZooKeeper),
			ClassLoader.getSystemClassLoader())).retrieveState();
	assertEquals(state, actual);
}