DiagnosticsUtil { public static Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> buildDiagnosticsManager(String jobName, String jobId, JobModel jobModel, String containerId, Optional<String> execEnvContainerId, Config config) { JobConfig jobConfig = new JobConfig(config); MetricsConfig metricsConfig = new MetricsConfig(config); Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> diagnosticsManagerReporterPair = Optional.empty(); if (jobConfig.getDiagnosticsEnabled()) { String diagnosticsReporterName = MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS; String diagnosticsFactoryClassName = metricsConfig.getMetricsFactoryClass(diagnosticsReporterName) .orElseThrow(() -> new SamzaException( String.format("Diagnostics reporter %s missing .class config", diagnosticsReporterName))); MetricsReporterFactory metricsReporterFactory = ReflectionUtil.getObj(diagnosticsFactoryClassName, MetricsReporterFactory.class); MetricsSnapshotReporter diagnosticsReporter = (MetricsSnapshotReporter) metricsReporterFactory.getMetricsReporter(diagnosticsReporterName, "samza-container-" + containerId, config); ClusterManagerConfig clusterManagerConfig = new ClusterManagerConfig(config); int containerMemoryMb = clusterManagerConfig.getContainerMemoryMb(); int containerNumCores = clusterManagerConfig.getNumCores(); long maxHeapSizeBytes = Runtime.getRuntime().maxMemory(); int containerThreadPoolSize = jobConfig.getThreadPoolSize(); String taskClassVersion = Util.getTaskClassVersion(config); String samzaVersion = Util.getSamzaVersion(); String hostName = Util.getLocalHost().getHostName(); Optional<String> diagnosticsReporterStreamName = metricsConfig.getMetricsSnapshotReporterStream(diagnosticsReporterName); if (!diagnosticsReporterStreamName.isPresent()) { throw new ConfigException( "Missing required config: " + String.format(MetricsConfig.METRICS_SNAPSHOT_REPORTER_STREAM, diagnosticsReporterName)); } SystemStream diagnosticsSystemStream = StreamUtil.getSystemStreamFromNames(diagnosticsReporterStreamName.get()); Optional<String> diagnosticsSystemFactoryName = new SystemConfig(config).getSystemFactory(diagnosticsSystemStream.getSystem()); if (!diagnosticsSystemFactoryName.isPresent()) { throw new SamzaException("Missing factory in config for system " + diagnosticsSystemStream.getSystem()); } SystemFactory systemFactory = ReflectionUtil.getObj(diagnosticsSystemFactoryName.get(), SystemFactory.class); SystemProducer systemProducer = systemFactory.getProducer(diagnosticsSystemStream.getSystem(), config, new MetricsRegistryMap(), DiagnosticsUtil.class.getSimpleName()); DiagnosticsManager diagnosticsManager = new DiagnosticsManager(jobName, jobId, jobModel.getContainers(), containerMemoryMb, containerNumCores, new StorageConfig(config).getNumPersistentStores(), maxHeapSizeBytes, containerThreadPoolSize, containerId, execEnvContainerId.orElse(""), taskClassVersion, samzaVersion, hostName, diagnosticsSystemStream, systemProducer, Duration.ofMillis(new TaskConfig(config).getShutdownMs()), jobConfig.getAutosizingEnabled(), config); diagnosticsManagerReporterPair = Optional.of(new ImmutablePair<>(diagnosticsManager, diagnosticsReporter)); } return diagnosticsManagerReporterPair; }  static void writeMetadataFile(String jobName, String jobId, String containerId,
      Optional<String> execEnvContainerId, Config config); static Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> buildDiagnosticsManager(String jobName,
      String jobId, JobModel jobModel, String containerId, Optional<String> execEnvContainerId, Config config); static void createDiagnosticsStream(Config config);  }
@Test public void testBuildDiagnosticsManagerReturnsConfiguredReporter() { Config config = new MapConfig(buildTestConfigs()); JobModel mockJobModel = mock(JobModel.class); SystemFactory systemFactory = mock(SystemFactory.class); SystemProducer mockProducer = mock(SystemProducer.class); MetricsReporterFactory metricsReporterFactory = mock(MetricsReporterFactory.class); MetricsSnapshotReporter mockReporter = mock(MetricsSnapshotReporter.class); when(systemFactory.getProducer(anyString(), any(Config.class), any(MetricsRegistry.class), anyString())).thenReturn(mockProducer); when(metricsReporterFactory.getMetricsReporter(anyString(), anyString(), any(Config.class))).thenReturn( mockReporter); PowerMockito.mockStatic(ReflectionUtil.class); when(ReflectionUtil.getObj(REPORTER_FACTORY, MetricsReporterFactory.class)).thenReturn(metricsReporterFactory); when(ReflectionUtil.getObj(SYSTEM_FACTORY, SystemFactory.class)).thenReturn(systemFactory); Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> managerReporterPair = DiagnosticsUtil.buildDiagnosticsManager(JOB_NAME, JOB_ID, mockJobModel, CONTAINER_ID, Optional.of(ENV_ID), config); Assert.assertTrue(managerReporterPair.isPresent()); Assert.assertEquals(mockReporter, managerReporterPair.get().getValue()); }