private boolean isValidRegisteredProcessor(final ProcessorNode processor) {
    String processorId = processor.getProcessorData().getProcessorId();
    List<ProcessorNode> processorNodes = getAllProcessorNodes().stream()
                                                               .filter(processorNode -> processorNode.processorData.getProcessorId().equals(processorId))
                                                               .collect(Collectors.toList());
    // Check for duplicate processor condition(if more than one processor exist for this processorId).
    if (processorNodes.size() > 1) {
      // There exists more than processor for provided `processorId`.
      LOG.debug("Processor nodes in zookeeper: {} for processorId: {}.", processorNodes, processorId);
      // Get all ephemeral processor paths
      TreeSet<String> sortedProcessorPaths = processorNodes.stream()
                                                           .map(ProcessorNode::getEphemeralPath)
                                                           .collect(Collectors.toCollection(TreeSet::new));
      // Check if smallest path is equal to this processor's ephemeralPath.
      return sortedProcessorPaths.first().equals(processor.getEphemeralPath());
    }
    // There're no duplicate processors. This is a valid registered processor.
    return true;
  }
Determines the validity of processor registered with zookeeper.

If there are multiple processors registered with same processorId,
the processor with lexicographically smallest zookeeperPath is considered valid
and all the remaining processors are invalid.

Two processors will not have smallest zookeeperPath because of sequentialId guarantees
of zookeeper for ephemeral nodes.

@param processor to check for validity condition in processors group.
@return true if the processor is valid. false otherwise.
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java