@Test public void testAddAndLock() throws Exception { LongStateStorage longStateStorage = new LongStateStorage(); ZooKeeperStateHandleStore<Long> store = new ZooKeeperStateHandleStore<>( ZOOKEEPER.getClient(), longStateStorage); final String pathInZooKeeper = "/testAdd"; final Long state = 1239712317L; store.addAndLock(pathInZooKeeper, state); assertEquals(1, store.getAllAndLock().size()); assertEquals(state, store.getAndLock(pathInZooKeeper).retrieveState()); Stat stat = ZOOKEEPER.getClient().checkExists().forPath(pathInZooKeeper); assertNotNull(stat); assertEquals(0, stat.getEphemeralOwner()); List<String> children = ZOOKEEPER.getClient().getChildren().forPath(pathInZooKeeper); assertEquals(1, children.size()); stat = ZOOKEEPER.getClient().checkExists().forPath(pathInZooKeeper + '/' + children.get(0)); assertNotNull(stat); assertNotEquals(0, stat.getEphemeralOwner()); @SuppressWarnings("unchecked") Long actual = ((RetrievableStateHandle<Long>) InstantiationUtil.deserializeObject( ZOOKEEPER.getClient().getData().forPath(pathInZooKeeper), ClassLoader.getSystemClassLoader())).retrieveState(); assertEquals(state, actual); }
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 { byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle); 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) { if (storeHandle != null) { storeHandle.discardState(); } } } }
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 { byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle); 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) { if (storeHandle != null) { storeHandle.discardState(); } } } } }
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 { byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle); 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) { if (storeHandle != null) { storeHandle.discardState(); } } } } ZooKeeperStateHandleStore(
		CuratorFramework client,
		RetrievableStateStorageHelper<T> storage); }
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 { byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle); 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) { if (storeHandle != null) { storeHandle.discardState(); } } } } ZooKeeperStateHandleStore(
		CuratorFramework client,
		RetrievableStateStorageHelper<T> storage); RetrievableStateHandle<T> addAndLock(
			String pathInZooKeeper,
			T state); void replace(String pathInZooKeeper, int expectedVersion, T state); int exists(String pathInZooKeeper); RetrievableStateHandle<T> getAndLock(String pathInZooKeeper); Collection<String> getAllPaths(); @SuppressWarnings("unchecked") List<Tuple2<RetrievableStateHandle<T>, String>> getAllAndLock(); @Nullable boolean releaseAndTryRemove(String pathInZooKeeper); void releaseAndTryRemoveAll(); void release(String pathInZooKeeper); void releaseAll(); void deleteChildren(); }
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 { byte[] serializedStoreHandle = InstantiationUtil.serializeObject(storeHandle); 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) { if (storeHandle != null) { storeHandle.discardState(); } } } } ZooKeeperStateHandleStore(
		CuratorFramework client,
		RetrievableStateStorageHelper<T> storage); RetrievableStateHandle<T> addAndLock(
			String pathInZooKeeper,
			T state); void replace(String pathInZooKeeper, int expectedVersion, T state); int exists(String pathInZooKeeper); RetrievableStateHandle<T> getAndLock(String pathInZooKeeper); Collection<String> getAllPaths(); @SuppressWarnings("unchecked") List<Tuple2<RetrievableStateHandle<T>, String>> getAllAndLock(); @Nullable boolean releaseAndTryRemove(String pathInZooKeeper); void releaseAndTryRemoveAll(); void release(String pathInZooKeeper); void releaseAll(); void deleteChildren();  }