ZkUtils { public void cleanupZK(int numVersionsToLeave) { deleteOldBarrierVersions(numVersionsToLeave); deleteOldJobModels(numVersionsToLeave); } ZkUtils(ZkKeyBuilder zkKeyBuilder, ZkClient zkClient, int connectionTimeoutMs, int sessionTimeOutMs, MetricsRegistry metricsRegistry); void incGeneration(); int getGeneration(); void connect(); synchronized void unregister(); ZkClient getZkClient(); ZkKeyBuilder getKeyBuilder(); synchronized String registerProcessorAndGetId(final ProcessorData data); void timeDelay(int sleepTimeInMillis); synchronized void deleteProcessorNode(String processorId); void writeTaskLocality(TaskName taskName, LocationId locationId); Map<TaskName, LocationId> readTaskLocality(); List<String> getSortedActiveProcessorsZnodes(); List<String> getSortedActiveProcessorsIDs(); List<String> getActiveProcessorsIDs(List<String> znodeIds); void unsubscribeDataChanges(String path, IZkDataListener dataListener); void subscribeDataChanges(String path, IZkDataListener dataListener); void subscribeChildChanges(String path, IZkChildListener listener); void unsubscribeChildChanges(String path, IZkChildListener childListener); void writeData(String path, Object object); boolean exists(String path); void close(); void subscribeToJobModelVersionChange(GenerationAwareZkDataListener dataListener); void publishJobModel(String jobModelVersion, JobModel jobModel); JobModel getJobModel(String jobModelVersion); String getJobModelVersion(); String getNextJobModelVersion(String currentJobModelVersion); void publishJobModelVersion(String oldVersion, String newVersion); void validateZkVersion(); void validatePaths(String[] paths); void subscribeToProcessorChange(IZkChildListener listener); void cleanupZK(int numVersionsToLeave);  }
@Test public void testCleanUpZk() { String pathA = "/path/testA"; String pathB = "/path/testB"; zkUtils.getZkClient().createPersistent(pathA, true); zkUtils.getZkClient().createPersistent(pathB, true); for (int i = 0; i < 20; i++) { String p1 = pathA + "/" + i; zkUtils.getZkClient().createPersistent(p1, true); zkUtils.getZkClient().createPersistent(p1 + "/something1", true); zkUtils.getZkClient().createPersistent(p1 + "/something2", true); String p2 = pathB + "/some_" + i; zkUtils.getZkClient().createPersistent(p2, true); zkUtils.getZkClient().createPersistent(p2 + "/something1", true); zkUtils.getZkClient().createPersistent(p2 + "/something2", true); } List<String> zNodeIds = new ArrayList<>(); zkUtils.deleteOldVersionPath(pathA, zNodeIds, 10, new Comparator<String>() { @Override public int compare(String o1, String o2) { return o1.compareTo(o2); } }); zNodeIds = zkUtils.getZkClient().getChildren(pathA); zkUtils.deleteOldVersionPath(pathA, zNodeIds, 10, new Comparator<String>() { @Override public int compare(String o1, String o2) { return Integer.valueOf(o1) - Integer.valueOf(o2); } }); for (int i = 0; i < 10; i++) { String p1 = pathA + "/" + i; Assert.assertFalse("path " + p1 + " exists", zkUtils.getZkClient().exists(p1)); } for (int i = 10; i < 20; i++) { String p1 = pathA + "/" + i; Assert.assertTrue("path " + p1 + " exists", zkUtils.getZkClient().exists(p1)); } zNodeIds = zkUtils.getZkClient().getChildren(pathB); zkUtils.deleteOldVersionPath(pathB, zNodeIds, 1, new Comparator<String>() { @Override public int compare(String o1, String o2) { return Integer.valueOf(o1.substring(o1.lastIndexOf("_") + 1)) - Integer .valueOf(o2.substring(o2.lastIndexOf("_") + 1)); } }); for (int i = 0; i < 19; i++) { String p1 = pathB + "/" + i; Assert.assertFalse("path " + p1 + " exists", zkUtils.getZkClient().exists(p1)); } for (int i = 19; i < 20; i++) { String p1 = pathB + "/some_" + i; Assert.assertTrue("path " + p1 + " exists", zkUtils.getZkClient().exists(p1)); } }