ZkUtils { public void subscribeToJobModelVersionChange(GenerationAwareZkDataListener dataListener) { LOG.info(" subscribing for jm version change at:" + keyBuilder.getJobModelVersionPath()); zkClient.subscribeDataChanges(keyBuilder.getJobModelVersionPath(), dataListener); metrics.subscriptions.inc(); } 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 testSubscribeToJobModelVersionChange() { ZkKeyBuilder keyBuilder = new ZkKeyBuilder("test"); String root = keyBuilder.getRootPath(); zkClient.deleteRecursive(root); class Result { String res = ""; public String getRes() { return res; } public void updateRes(String newRes) { res = newRes; } } Assert.assertFalse(zkUtils.exists(root)); zkUtils.validatePaths(new String[]{root, keyBuilder.getJobModelVersionPath(), keyBuilder.getProcessorsPath()}); Assert.assertTrue(zkUtils.exists(root)); Assert.assertTrue(zkUtils.exists(keyBuilder.getJobModelVersionPath())); Assert.assertTrue(zkUtils.exists(keyBuilder.getProcessorsPath())); final Result res = new Result(); IZkDataListener dataListener = new IZkDataListener() { @Override public void handleDataChange(String dataPath, Object data) throws Exception { res.updateRes((String) data); } @Override public void handleDataDeleted(String dataPath) throws Exception { Assert.fail("Data wasn't deleted;"); } }; zkClient.subscribeDataChanges(keyBuilder.getJobModelVersionPath(), dataListener); zkClient.subscribeDataChanges(keyBuilder.getProcessorsPath(), dataListener); zkClient.writeData(keyBuilder.getJobModelVersionPath(), "newVersion"); Assert.assertTrue(testWithDelayBackOff(() -> "newVersion".equals(res.getRes()), 2, 1000)); zkClient.writeData(keyBuilder.getProcessorsPath(), "newProcessor"); Assert.assertTrue(testWithDelayBackOff(() -> "newProcessor".equals(res.getRes()), 2, 1000)); }