ZkUtils { List<ProcessorNode> getAllProcessorNodes() { List<String> processorZNodes = getSortedActiveProcessorsZnodes(); LOG.debug("Active ProcessorZNodes in zookeeper: {}.", processorZNodes); List<ProcessorNode> processorNodes = new ArrayList<>(); for (String processorZNode: processorZNodes) { String ephemeralProcessorPath = String.format("%s/%s", keyBuilder.getProcessorsPath(), processorZNode); String data = readProcessorData(ephemeralProcessorPath); if (data != null) { processorNodes.add(new ProcessorNode(new ProcessorData(data), ephemeralProcessorPath)); } } return processorNodes; } 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 testGetAllProcessorNodesShouldReturnEmptyForNonExistingZookeeperNodes() { List<ZkUtils.ProcessorNode> processorsIDs = zkUtils.getAllProcessorNodes(); Assert.assertEquals(0, processorsIDs.size()); }