RemoteApplicationRunner implements ApplicationRunner { @Override public void run(ExternalContext externalContext) { if (new JobConfig(config).getConfigLoaderFactory().isPresent()) { JobRunner runner = new JobRunner(JobPlanner.generateSingleJobConfig(config)); runner.submit(); return; } try { JobPlanner planner = new RemoteJobPlanner(ApplicationDescriptorUtil.getAppDescriptor(app, config)); List<JobConfig> jobConfigs = planner.prepareJobs(); if (jobConfigs.isEmpty()) { throw new SamzaException("No jobs to run."); } 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); } } RemoteApplicationRunner(SamzaApplication app, Config config); @Override void run(ExternalContext externalContext); @Override void kill(); @Override ApplicationStatus status(); @Override void waitForFinish(); @Override boolean waitForFinish(Duration timeout);  }
@Test public void testRunWithConfigLoaderFactoryPresent() { Map<String, String> config = new HashMap<>(); config.put(ApplicationConfig.APP_NAME, "test-app"); config.put(JobConfig.CONFIG_LOADER_FACTORY, "org.apache.samza.config.loaders.PropertiesConfigLoaderFactory"); config.put(JobConfig.STREAM_JOB_FACTORY_CLASS, MockStreamJobFactory.class.getName()); runner = new RemoteApplicationRunner(null, new MapConfig(config)); runner.run(null); }