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);
        // Determine if there are duplicate processors with this.processorId after registration.
        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;
  }
Returns a ZK generated identifier for this client.
If the current client is registering for the first time, it creates an ephemeral sequential node in the ZK tree
If the current client has already registered and is still within the same session, it returns the already existing
value for the ephemeralPath

@param data Object that should be written as data in the registered ephemeral ZK node
@return String representing the absolute ephemeralPath of this client in the current session
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java