ZkUtils { public synchronized String registerProcessorAndGetId(final ProcessorData data) { final long startTimeMs = System.currentTimeMillis(); final long retryTimeOutMs = 2 * sessionTimeoutMs; String processorId = data.getProcessorId(); if (ephemeralPath == null) { ephemeralPath = zkClient.createEphemeralSequential(keyBuilder.getProcessorsPath() + "/", data.toString()); LOG.info("Created ephemeral path: {} for processor: {} in zookeeper.", ephemeralPath, data); while (true) { ProcessorNode processorNode = new ProcessorNode(data, ephemeralPath); if (!isValidRegisteredProcessor(processorNode)) { long currentTimeMs = System.currentTimeMillis(); if ((currentTimeMs - startTimeMs) < retryTimeOutMs) { LOG.info("Processor: {} is duplicate. Retrying registration again.", processorId); timeDelay(1000); } else { LOG.info("Processor: {} is duplicate. Deleting zookeeper node at path: {}.", processorId, ephemeralPath); zkClient.delete(ephemeralPath); metrics.deletions.inc(); throw new SamzaException(String.format("Processor: %s is duplicate in the group. Registration failed.", processorId)); } } else { break; } } } else { LOG.info("Ephemeral path: {} exists for processor: {} in zookeeper.", ephemeralPath, data); } return ephemeralPath; } 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 testRegisterProcessorId() { String assignedPath = zkUtils.registerProcessorAndGetId(new ProcessorData("host", "1")); Assert.assertTrue(assignedPath.startsWith(KEY_BUILDER.getProcessorsPath())); Assert.assertTrue(zkUtils.registerProcessorAndGetId(new ProcessorData("host", "1")).equals(assignedPath)); }