ZkUtils { public List<String> getActiveProcessorsIDs(List<String> znodeIds) { String processorPath = keyBuilder.getProcessorsPath(); List<String> processorIds = new ArrayList<>(znodeIds.size()); if (znodeIds.size() > 0) { for (String child : znodeIds) { String fullPath = String.format("%s/%s", processorPath, child); String processorData = readProcessorData(fullPath); if (processorData != null) { processorIds.add(new ProcessorData(processorData).getProcessorId()); } } Collections.sort(processorIds); LOG.info("Found these children - " + znodeIds); LOG.info("Found these processorIds - " + processorIds); } return processorIds; } 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 testGetActiveProcessorIdShouldReturnEmptyForNonExistingZookeeperNodes() { List<String> processorsIDs = zkUtils.getActiveProcessorsIDs(ImmutableList.of("node1", "node2")); Assert.assertEquals(0, processorsIDs.size()); }