JobCoordinatorLaunchUtil { @SuppressWarnings("rawtypes") public static void run(SamzaApplication app, Config config) { ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc = ApplicationDescriptorUtil.getAppDescriptor(app, config); RemoteJobPlanner planner = new RemoteJobPlanner(appDesc); List<JobConfig> jobConfigs = planner.prepareJobs(); if (jobConfigs.size() != 1) { throw new SamzaException("Only support single remote job is supported."); } Config fullConfig = jobConfigs.get(0); CoordinatorStreamUtil.createCoordinatorStream(fullConfig); MetricsRegistryMap metrics = new MetricsRegistryMap(); MetadataStore metadataStore = new CoordinatorStreamStore(CoordinatorStreamUtil.buildCoordinatorStreamConfig(fullConfig), metrics); metadataStore.init(); Config launchConfig = CoordinatorStreamUtil.readLaunchConfigFromCoordinatorStream(fullConfig, metadataStore); Config finalConfig = new MapConfig(launchConfig, fullConfig); CoordinatorStreamUtil.writeConfigToCoordinatorStream(finalConfig, true); DiagnosticsUtil.createDiagnosticsStream(finalConfig); ClusterBasedJobCoordinator jc = new ClusterBasedJobCoordinator( metrics, metadataStore, finalConfig); jc.run(); } private  JobCoordinatorLaunchUtil(); @SuppressWarnings("rawtypes") static void run(SamzaApplication app, Config config);  }
@Test public void testCreateFromConfigLoader() throws Exception { Map<String, String> config = new HashMap<>(); config.put(JobConfig.CONFIG_LOADER_FACTORY, PropertiesConfigLoaderFactory.class.getName()); config.put(PropertiesConfigLoaderFactory.CONFIG_LOADER_PROPERTIES_PREFIX + "path", getClass().getResource("/test.properties").getPath()); Config originalConfig = new JobConfig(ConfigUtil.loadConfig(new MapConfig(config))); Config fullConfig = new MapConfig(originalConfig, Collections.singletonMap("isAfterPlanning", "true")); Config autoSizingConfig = new MapConfig(Collections.singletonMap(JobConfig.JOB_AUTOSIZING_CONTAINER_COUNT, "10")); Config finalConfig = new MapConfig(autoSizingConfig, fullConfig); RemoteJobPlanner mockJobPlanner = mock(RemoteJobPlanner.class); CoordinatorStreamStore mockCoordinatorStreamStore = mock(CoordinatorStreamStore.class); ClusterBasedJobCoordinator mockJC = mock(ClusterBasedJobCoordinator.class); PowerMockito.mockStatic(CoordinatorStreamUtil.class); PowerMockito.doNothing().when(CoordinatorStreamUtil.class, "createCoordinatorStream", any()); PowerMockito.doReturn(new MapConfig()).when(CoordinatorStreamUtil.class, "buildCoordinatorStreamConfig", any()); PowerMockito.doReturn(autoSizingConfig).when(CoordinatorStreamUtil.class, "readLaunchConfigFromCoordinatorStream", any(), any()); PowerMockito.whenNew(CoordinatorStreamStore.class).withAnyArguments().thenReturn(mockCoordinatorStreamStore); PowerMockito.whenNew(RemoteJobPlanner.class).withAnyArguments().thenReturn(mockJobPlanner); PowerMockito.whenNew(ClusterBasedJobCoordinator.class).withAnyArguments().thenReturn(mockJC); when(mockJobPlanner.prepareJobs()).thenReturn(Collections.singletonList(new JobConfig(fullConfig))); JobCoordinatorLaunchUtil.run(new MockStreamApplication(), originalConfig); verifyNew(ClusterBasedJobCoordinator.class).withArguments(any(MetricsRegistryMap.class), eq(mockCoordinatorStreamStore), eq(finalConfig)); verify(mockJC, times(1)).run(); verifyStatic(times(1)); CoordinatorStreamUtil.createCoordinatorStream(any()); verifyStatic(times(1)); CoordinatorStreamUtil.writeConfigToCoordinatorStream(any(), anyBoolean()); }