public static Optional<DiagnosticsManager> buildDiagnosticsManager(String jobName, String jobId, JobModel jobModel,
      String containerId, Optional<String> executionEnvContainerId, Optional<String> samzaEpochId, Config config) {

    JobConfig jobConfig = new JobConfig(config);
    MetricsConfig metricsConfig = new MetricsConfig(config);
    Optional<DiagnosticsManager> diagnosticsManagerOptional = Optional.empty();

    if (jobConfig.getDiagnosticsEnabled()) {
      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(MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS);

      if (!diagnosticsReporterStreamName.isPresent()) {
        throw new ConfigException(
            "Missing required config: " + String.format(MetricsConfig.METRICS_SNAPSHOT_REPORTER_STREAM,
                MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS));
      }
      SystemStream diagnosticsSystemStream = StreamUtil.getSystemStreamFromNames(diagnosticsReporterStreamName.get());

      // Create a SystemProducer for DiagnosticsManager. This producer is used by the DiagnosticsManager
      // to write to the same stream as the MetricsSnapshotReporter called `diagnosticsreporter`.
      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, executionEnvContainerId.orElse(""), samzaEpochId.orElse(""), taskClassVersion,
              samzaVersion, hostName, diagnosticsSystemStream, systemProducer,
              Duration.ofMillis(new TaskConfig(config).getShutdownMs()), jobConfig.getAutosizingEnabled(), config);

      diagnosticsManagerOptional = Optional.of(diagnosticsManager);
    }

    return diagnosticsManagerOptional;
  }
Create a {@link DiagnosticsManager} for the given jobName, jobId, containerId, and execEnvContainerId, if
diagnostics is enabled.
@param executionEnvContainerId ID assigned to the container by the cluster manager (e.g. YARN)
@param samzaEpochId ID assigned to the job deployment attempt by the cluster manager
samza-core/src/main/java/org/apache/samza/util/DiagnosticsUtil.java