@Override
  public List<JobConfig> prepareJobs() {
    // for high-level DAG, generate the plan and job configs
    // run.id is based on current system time with the most significant bits in UUID (8 digits) to avoid collision
    String runId = String.valueOf(System.currentTimeMillis()) + "-" + UUID.randomUUID().toString().substring(0, 8);
    LOG.info("The run id for this run is {}", runId);

    // 1. initialize and plan
    ExecutionPlan plan = getExecutionPlan(runId);
    try {
      writePlanJsonFile(plan.getPlanAsJson());
    } catch (Exception e) {
      throw new SamzaException("Failed to create plan JSON.", e);
    }

    List<JobConfig> jobConfigs = plan.getJobConfigs();
    if (jobConfigs.isEmpty()) {
      throw new SamzaException("No jobs in the plan.");
    }

    // 2. create the necessary streams
    // TODO: this works for single-job applications. For multi-job applications, ExecutionPlan should return an AppConfig
    // to be used for the whole application
    JobConfig jobConfig = jobConfigs.get(0);
    StreamManager streamManager = null;
    try {
      // create the StreamManager to create intermediate streams in the plan
      streamManager = buildAndStartStreamManager(jobConfig);
      if (plan.getApplicationConfig().getAppMode() == ApplicationConfig.ApplicationMode.BATCH) {
        streamManager.clearStreamsFromPreviousRun(getConfigFromPrevRun());
      }
      streamManager.createStreams(plan.getIntermediateStreams());
    } finally {
      if (streamManager != null) {
        streamManager.stop();
      }
    }
    return jobConfigs;
  }
Temporary helper class with specific implementation of {@link JobPlanner#prepareJobs()}
for remote-launched Samza processors (e.g. in YARN).

TODO: we need to consolidate this class with {@link ExecutionPlanner} after SAMZA-1811.
samza-core/src/main/java/org/apache/samza/execution/RemoteJobPlanner.java