ContainerProcessManager implements ClusterResourceManager.Callback { public ContainerProcessManager(Config config, SamzaApplicationState state, MetricsRegistryMap registry, ContainerPlacementMetadataStore metadataStore, LocalityManager localityManager) { Preconditions.checkNotNull(localityManager, "Locality manager cannot be null"); this.state = state; this.clusterManagerConfig = new ClusterManagerConfig(config); this.jobConfig = new JobConfig(config); this.hostAffinityEnabled = clusterManagerConfig.getHostAffinityEnabled(); ResourceManagerFactory factory = getContainerProcessManagerFactory(clusterManagerConfig); this.clusterResourceManager = checkNotNull(factory.getClusterResourceManager(this, state)); this.containerProcessManagerMetrics = new ContainerProcessManagerMetrics(config, state, registry); this.jvmMetrics = new JvmMetrics(registry); this.metricsReporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), METRICS_SOURCE_NAME); String jobName = new JobConfig(config).getName().get(); String jobId = new JobConfig(config).getJobId(); Optional<String> execEnvContainerId = Optional.ofNullable(System.getenv(EXEC_ENV_CONTAINER_ID_SYS_PROPERTY)); Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> diagnosticsManagerReporterPair = DiagnosticsUtil.buildDiagnosticsManager(jobName, jobId, state.jobModelManager.jobModel(), METRICS_SOURCE_NAME, execEnvContainerId, config); if (diagnosticsManagerReporterPair.isPresent()) { diagnosticsManager = Option.apply(diagnosticsManagerReporterPair.get().getKey()); metricsReporters.put(MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS, diagnosticsManagerReporterPair.get().getValue()); } else { diagnosticsManager = Option.empty(); } this.localityManager = localityManager; this.metricsReporters.values().forEach(reporter -> reporter.register(METRICS_SOURCE_NAME, registry)); this.containerManager = new ContainerManager(metadataStore, state, clusterResourceManager, hostAffinityEnabled, jobConfig.getStandbyTasksEnabled(), localityManager); this.containerAllocator = new ContainerAllocator(this.clusterResourceManager, config, state, hostAffinityEnabled, this.containerManager); this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread"); LOG.info("Finished container process manager initialization."); } ContainerProcessManager(Config config, SamzaApplicationState state, MetricsRegistryMap registry,
      ContainerPlacementMetadataStore metadataStore, LocalityManager localityManager); @VisibleForTesting  ContainerProcessManager(ClusterManagerConfig clusterManagerConfig,
      SamzaApplicationState state,
      MetricsRegistryMap registry,
      ClusterResourceManager resourceManager,
      Optional<ContainerAllocator> allocator,
      ContainerManager containerManager,
      LocalityManager localityManager); boolean shouldShutdown(); void start(); void stop(); void onResourceAllocated(SamzaResource resource); void onResourceCompleted(SamzaResourceStatus resourceStatus); @Override void onResourcesAvailable(List<SamzaResource> resources); @Override void onResourcesCompleted(List<SamzaResourceStatus> resourceStatuses); @Override void onStreamProcessorLaunchSuccess(SamzaResource resource); @Override void onStreamProcessorLaunchFailure(SamzaResource resource, Throwable t); @Override void onStreamProcessorStopFailure(SamzaResource resource, Throwable t); @Override void onError(Throwable e); void registerContainerPlacementAction(ContainerPlacementRequestMessage requestMessage);  }
@Test public void testContainerProcessManager() throws Exception { Map<String, String> conf = new HashMap<>(); conf.putAll(getConfig()); conf.put("cluster-manager.container.memory.mb", "500"); conf.put("cluster-manager.container.cpu.cores", "5"); SamzaApplicationState state = new SamzaApplicationState(getJobModelManager(1)); MockClusterResourceManagerCallback callback = new MockClusterResourceManagerCallback(); MockClusterResourceManager clusterResourceManager = new MockClusterResourceManager(callback, state); LocalityManager mockLocalityManager = mock(LocalityManager.class); when(mockLocalityManager.readLocality()) .thenReturn(new LocalityModel(ImmutableMap.of("0", new ProcessorLocality("0", "host1")))); ContainerManager containerManager = buildContainerManager(containerPlacementMetadataStore, state, clusterResourceManager, true, false, mockLocalityManager); ContainerProcessManager cpm = buildContainerProcessManager(new ClusterManagerConfig(new MapConfig(conf)), state, clusterResourceManager, Optional.empty()); ContainerAllocator allocator = (ContainerAllocator) getPrivateFieldFromCpm("containerAllocator", cpm).get(cpm); assertEquals(ContainerAllocator.class, allocator.getClass()); assertEquals(500, allocator.containerMemoryMb); assertEquals(5, allocator.containerNumCpuCores); conf.clear(); conf.putAll(getConfigWithHostAffinity()); conf.put("cluster-manager.container.memory.mb", "500"); conf.put("cluster-manager.container.cpu.cores", "5"); state = new SamzaApplicationState(getJobModelManager(1)); callback = new MockClusterResourceManagerCallback(); clusterResourceManager = new MockClusterResourceManager(callback, state); cpm = new ContainerProcessManager( new ClusterManagerConfig(new MapConfig(conf)), state, new MetricsRegistryMap(), clusterResourceManager, Optional.empty(), containerManager, mockLocalityManager ); allocator = (ContainerAllocator) getPrivateFieldFromCpm("containerAllocator", cpm).get(cpm); assertEquals(ContainerAllocator.class, allocator.getClass()); assertEquals(500, allocator.containerMemoryMb); assertEquals(5, allocator.containerNumCpuCores); }