@Override
  public List<JobConfig> prepareJobs() {
    // for high-level DAG, generating the plan and job configs
    // 1. initialize and plan
    ExecutionPlan plan = getExecutionPlan(runId);

    String executionPlanJson = "";
    try {
      executionPlanJson = plan.getPlanAsJson();
    } catch (Exception e) {
      throw new SamzaException("Failed to create plan JSON.", e);
    }
    writePlanJsonFile(executionPlanJson);
    LOG.info("Execution Plan: \n" + executionPlanJson);
    String planId = String.valueOf(executionPlanJson.hashCode());

    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);
      createStreams(planId, plan.getIntermediateStreams(), streamManager);
    } finally {
      if (streamManager != null) {
        streamManager.stop();
      }
    }
    return jobConfigs;
  }
Temporarily helper class with specific implementation of {@link JobPlanner#prepareJobs()}
for standalone Samza processors.

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