ClusterBasedJobCoordinatorRunner { @VisibleForTesting static void runClusterBasedJobCoordinator(String[] args) { final String coordinatorSystemEnv = System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG); final String submissionEnv = System.getenv(ShellCommandConfig.ENV_SUBMISSION_CONFIG); if (!StringUtils.isBlank(submissionEnv)) { Config submissionConfig; try { LOG.info("Parsing submission config {}", submissionEnv); submissionConfig = new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(submissionEnv, Config.class)); LOG.info("Using the submission config: {}.", submissionConfig); } catch (IOException e) { LOG.error("Exception while reading submission config", e); throw new SamzaException(e); } ApplicationConfig appConfig = new ApplicationConfig(submissionConfig); String className = appConfig.getAppMainClass(); String[] arguments = toArgs(appConfig); LOG.info("Invoke main {} with args {}", className, arguments); try { Class<?> cls = Class.forName(className); Method mainMethod = cls.getMethod("main", String[].class); mainMethod.invoke(null, (Object) arguments); } catch (Exception e) { throw new SamzaException(e); } } else { Config coordinatorSystemConfig; try { LOG.info("Parsing coordinator system config {}", coordinatorSystemEnv); coordinatorSystemConfig = new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(coordinatorSystemEnv, Config.class)); LOG.info("Using the coordinator system config: {}.", coordinatorSystemConfig); } catch (IOException e) { LOG.error("Exception while reading coordinator stream config", e); throw new SamzaException(e); } ClusterBasedJobCoordinator jc = createFromMetadataStore(coordinatorSystemConfig); jc.run(); } LOG.info("Finished running ClusterBasedJobCoordinator"); }  static void main(String[] args);  }
@Test public void testRunClusterBasedJobCoordinator() throws Exception { Config submissionConfig = new MapConfig(ImmutableMap.of( JobConfig.CONFIG_LOADER_FACTORY, PropertiesConfigLoaderFactory.class.getName(), PropertiesConfigLoaderFactory.CONFIG_LOADER_PROPERTIES_PREFIX + "path", getClass().getResource("/test.properties").getPath())); Config fullConfig = ConfigUtil.loadConfig(submissionConfig); StreamApplication mockApplication = mock(StreamApplication.class); PowerMockito.mockStatic(System.class, ApplicationUtil.class, JobCoordinatorLaunchUtil.class); PowerMockito .when(System.getenv(eq(ShellCommandConfig.ENV_SUBMISSION_CONFIG))) .thenReturn(SamzaObjectMapper.getObjectMapper().writeValueAsString(submissionConfig)); PowerMockito .when(ApplicationUtil.fromConfig(any())) .thenReturn(mockApplication); PowerMockito.doNothing().when(JobCoordinatorLaunchUtil.class, "run", mockApplication, fullConfig); ClusterBasedJobCoordinatorRunner.runClusterBasedJobCoordinator(null); PowerMockito.verifyStatic(times(1)); JobCoordinatorLaunchUtil.run(mockApplication, fullConfig); }