private void checkAndCreateStreams(String lockId, List<StreamSpec> intStreams, StreamManager streamManager) throws TimeoutException {
    MetadataStore metadataStore = getMetadataStore();
    DistributedLock distributedLock = coordinationUtils.getLock(lockId);
    if (distributedLock == null || metadataStore == null) {
      LOG.warn("Processor {} failed to create utils. Each processor will attempt to create streams.", processorId);
      // each application process will try creating the streams, which requires stream creation to be idempotent
      streamManager.createStreams(intStreams);
      return;
    }
    //Start timer for timeout
    long startTime = System.currentTimeMillis();
    long lockTimeout = TimeUnit.MILLISECONDS.convert(CoordinationConstants.LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS);

    // If "stream created state" exists in store then skip stream creation
    // Else acquire lock, create streams, set state in store and unlock
    // Checking for state before acquiring lock to prevent all processors from acquiring lock
    // In a while loop so that if two processors check state simultaneously then
    // to make sure the processor not acquiring the lock
    // does not die of timeout exception and comes back and checks for state and proceeds
    while ((System.currentTimeMillis() - startTime) < lockTimeout) {
      if (metadataStore.get(String.format(STREAM_CREATED_STATE_KEY, lockId)) != null) {
        LOG.info("Processor {} found streams created state data. They must've been created by another processor.", processorId);
        break;
      }
      try {
        if (distributedLock.lock(Duration.ofMillis(10000))) {
          LOG.info("lock acquired for streams creation by Processor " + processorId);
          streamManager.createStreams(intStreams);
          String streamCreatedMessage = "Streams created by processor " + processorId;
          metadataStore.put(String.format(STREAM_CREATED_STATE_KEY, lockId), streamCreatedMessage.getBytes("UTF-8"));
          metadataStore.flush();
          distributedLock.unlock();
          break;
        } else {
          LOG.info("Processor {} failed to get the lock for stream initialization. Will try again until time out", processorId);
        }
      } catch (UnsupportedEncodingException e) {
        String msg = String.format("Processor {} failed to encode string for stream initialization", processorId);
        throw new SamzaException(msg, e);
      }
    }
    if ((System.currentTimeMillis() - startTime) >= lockTimeout) {
      throw new TimeoutException(String.format("Processor {} failed to get the lock for stream initialization within {} milliseconds.", processorId, CoordinationConstants.LOCK_TIMEOUT_MS));
    }
  }
Create intermediate streams using {@link StreamManager}.
If {@link CoordinationUtils} is provided, this function will first invoke leader election, and the leader
will create the streams. All the runner processes will wait on the latch that is released after the leader finishes
stream creation.
@param planId a unique identifier representing the plan used for coordination purpose
@param intStreams list of intermediate {@link StreamSpec}s
@param streamManager the {@link StreamManager} used to create streams
samza-core/src/main/java/org/apache/samza/execution/LocalJobPlanner.java