@Override
  public void run(ExternalContext externalContext) {
    // By default with RemoteApplication runner we are going to defer the planning to JobCoordinatorRunner
    // with the exception of local deployment
    JobConfig userJobConfig = new JobConfig(config);
    if (userJobConfig.getConfigLoaderFactory().isPresent() && !isLocalDeployment(userJobConfig)) {
      JobRunner runner = new JobRunner(JobPlanner.generateSingleJobConfig(config));
      runner.submit();
      return;
    }

    // TODO SAMZA-2432: Clean this up once SAMZA-2405 is completed when legacy flow is removed.
    try {
      JobPlanner planner = getRemoteJobPlanner();
      List<JobConfig> jobConfigs = planner.prepareJobs();
      if (jobConfigs.isEmpty()) {
        throw new SamzaException("No jobs to run.");
      }

      // 3. submit jobs for remote execution
      jobConfigs.forEach(jobConfig -> {
        LOG.info("Starting job {} with config {}", jobConfig.getName(), jobConfig);
        JobRunner runner = new JobRunner(jobConfig);
        runner.run(true);
      });
    } catch (Throwable t) {
      throw new SamzaException("Failed to run application", t);
    }
  }
Constructors a {@link RemoteApplicationRunner} to run the {@code app} with the {@code config}.

@param app application to run
@param config configuration for the application
samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java