@Override
  public boolean lock(Duration timeout) {

    nodePath = zkUtils.getZkClient().createEphemeralSequential(lockPath + "/", participantId);

    //Start timer for timeout
    long startTime = System.currentTimeMillis();
    long lockTimeout = timeout.toMillis();

    while ((System.currentTimeMillis() - startTime) < lockTimeout) {
      synchronized (mutex) {
        List<String> children = zkUtils.getZkClient().getChildren(lockPath);
        int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(nodePath));

        if (children.size() == 0 || index == -1) {
          throw new SamzaException("Looks like we are no longer connected to Zk. Need to reconnect!");
        }
        // Acquires lock when the node has the lowest sequence number and returns.
        if (index == 0) {
          LOG.info("Acquired lock for participant id: {}", participantId);
          return true;
        } else {
          try {
            mutex.wait(lockTimeout);
          } catch (InterruptedException e) {
            Thread.interrupted();
          }
          LOG.info("Trying to acquire lock again...");
        }
      }
    }
    LOG.info("Failed to acquire lock within {} milliseconds.", lockTimeout);
    return false;
  }
Tries to acquire a lock in order to create intermediate streams. On failure to acquire lock, it keeps trying until the lock times out.
Creates a sequential ephemeral node to acquire the lock. If the path of this node has the lowest sequence number, the processor has acquired the lock.
@param timeout Duration of lock acquiring timeout.
@return true if lock is acquired successfully else returns false if failed to acquire within timeout
samza-core/src/main/java/org/apache/samza/zk/ZkDistributedLock.java