private void createStreams(String planId, List<StreamSpec> intStreams, StreamManager streamManager) {
    if (intStreams.isEmpty()) {
      LOG.info("Set of intermediate streams is empty. Nothing to create.");
      return;
    }
    LOG.info("A single processor must create the intermediate streams. Processor {} will attempt to acquire the lock.", processorId);
    // Move the scope of coordination utils within stream creation to address long idle connection problem.
    // Refer SAMZA-1385 for more details
    if (coordinationUtils == 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;
    }

    // If BATCH, then need to create new intermediate streams every run.
    // planId does not change every run and hence, need to use runid
    // as the lockId to create a new lock with state each run
    // to create new streams each run.
    // If run.id is null, defaults to old behavior of using planId
    boolean isAppModeBatch = new ApplicationConfig(userConfig).getAppMode() == ApplicationConfig.ApplicationMode.BATCH;
    String lockId = planId;
    if (isAppModeBatch && runId != null) {
      lockId = runId;
    }
    try {
      checkAndCreateStreams(lockId, intStreams, streamManager);
    } catch (TimeoutException te) {
      throw new SamzaException(String.format("Processor {} failed to get the lock for stream initialization within timeout.", processorId), te);
    } finally {
      if (!isAppModeBatch && coordinationUtils != null) {
        coordinationUtils.close();
      }
    }
  }
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