ZkUtils { public String getNextJobModelVersion(String currentJobModelVersion) { if (currentJobModelVersion == null) { return "1"; } else { List<String> publishedJobModelVersions = zkClient.getChildren(keyBuilder.getJobModelPathPrefix()); metrics.reads.inc(publishedJobModelVersions.size()); String maxPublishedJMVersion = publishedJobModelVersions.stream() .max(Comparator.comparingInt(Integer::valueOf)).orElse("0"); return Integer.toString(Math.max(Integer.valueOf(currentJobModelVersion), Integer.valueOf(maxPublishedJMVersion)) + 1); } } 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 testgetNextJobModelVersion() { ZkKeyBuilder keyBuilder = new ZkKeyBuilder("test"); String root = keyBuilder.getRootPath(); zkClient.deleteRecursive(root); zkUtils.validatePaths(new String[]{root, keyBuilder.getJobModelPathPrefix(), keyBuilder.getJobModelVersionPath()}); String version = "1"; String oldVersion = "0"; zkUtils.publishJobModelVersion(oldVersion, version); Assert.assertEquals(version, zkUtils.getJobModelVersion()); zkUtils.publishJobModel("2", new JobModel(new MapConfig(), new HashMap<>())); Assert.assertEquals("3", zkUtils.getNextJobModelVersion(zkUtils.getJobModelVersion())); }