ZkUtils { public static <T> void changeZkValue(CuratorFramework client, String path, Function<T, T> changeFunction, Function<byte[], T> decoder, Function<T, byte[]> encoder) { Function<byte[], byte[]> realFunction = old -> { T decodedOld = decoder.apply(old); return encoder.apply(changeFunction.apply(decodedOld)); }; changeZkValue(client, path, realFunction, INFINITY_LOOP, DEFAULT_WAIT); } private  ZkUtils(); static String getStringFromZk(CuratorFramework client, String path); static byte[] getBytesFromZk(CuratorFramework client, String path); static T getFromZk(CuratorFramework client, String path,
            ThrowableFunction<byte[], T, X> decoder); static void setToZk(CuratorFramework client, String path, byte[] value); @Deprecated static void setToZk(CuratorFramework client, String path, byte[] value,
            CreateMode createMode); static EphemeralNode createEphemeralNode(CuratorFramework client, String path,
            byte[] value); static void removeFromZk(CuratorFramework client, String path); static void removeFromZk(CuratorFramework client, String path,
            boolean recruitDeletedChildren); static void changeZkValue(CuratorFramework client, String path,
            Function<T, T> changeFunction, Function<byte[], T> decoder,
            Function<T, byte[]> encoder); static boolean changeZkValue(CuratorFramework client, String path,
            Function<byte[], byte[]> changeFunction, int retryTimes, long retryWait); static Stream<ChildData> getAllChildrenWithData(CuratorFramework curator, String parentPath); static Stream<String> getAllChildren(CuratorFramework curator, String parentPath);  }
@Test void testModify() throws Exception { ZkUtils.changeZkValue(curatorFramework, TEST_PATH, old -> { old.add("1"); return old; }, this::setDecode, this::setEncode); assertEquals(setDecode(curatorFramework.getData().forPath(TEST_PATH)), Sets.newHashSet("1")); ZkUtils.changeZkValue(curatorFramework, TEST_PATH, old -> { old.remove("1"); return old; }, this::setDecode, this::setEncode); assertEquals(setDecode(curatorFramework.getData().forPath(TEST_PATH)), Sets.newHashSet()); }
@Test void testConcurrent() throws Exception { ExecutorService executor = Executors.newFixedThreadPool(100); for (int i = 0; i < LOOP; i++) { executor.execute(() -> { ZkUtils.changeZkValue(curatorFramework, TEST_PATH2, old -> old + 1, this::intDecode, this::intEncode); }); } MoreExecutors.shutdownAndAwaitTermination(executor, 1, DAYS); assertEquals(intDecode(curatorFramework.getData().forPath(TEST_PATH2)), LOOP); }