@Test
public void testLockCleanupWhenClientTimesOut() throws Exception {

	LongStateStorage longStateStorage = new LongStateStorage();
	Configuration configuration = new Configuration();
	configuration.setString(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, ZOOKEEPER.getConnectString());
	configuration.setInteger(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT, 100);
	configuration.setString(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT, "timeout");
	try (CuratorFramework client = ZooKeeperUtils.startCuratorFramework(configuration);
		CuratorFramework client2 = ZooKeeperUtils.startCuratorFramework(configuration)) {
		ZooKeeperStateHandleStore<Long> zkStore = new ZooKeeperStateHandleStore<>(
			client,
			longStateStorage,
			Executors.directExecutor());
		final String path = "/state";
		zkStore.addAndLock(path, 42L);
		// this should delete all ephemeral nodes
		client.close();
		Stat stat = client2.checkExists().forPath(path);
		// check that our state node still exists
		assertNotNull(stat);
		Collection<String> children = client2.getChildren().forPath(path);
		// check that the lock node has been released
		assertEquals(0, children.size());
	}
}