ZkUtils { public Map<TaskName, LocationId> readTaskLocality() { Map<TaskName, LocationId> taskLocality = new HashMap<>(); String taskLocalityPath = keyBuilder.getTaskLocalityPath(); List<String> tasks = new ArrayList<>(); if (zkClient.exists(taskLocalityPath)) { tasks = zkClient.getChildren(taskLocalityPath); } for (String taskName : tasks) { String taskPath = String.format("%s/%s", keyBuilder.getTaskLocalityPath(), taskName); String locationId = zkClient.readData(taskPath, true); if (locationId != null) { taskLocality.put(new TaskName(taskName), new LocationId(locationId)); } } return taskLocality; } 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 testReadWhenTaskLocalityDoesNotExist() { Map<TaskName, LocationId> taskLocality = zkUtils.readTaskLocality(); Assert.assertEquals(0, taskLocality.size()); }