public static void createDiagnosticsStream(Config config) {
    if (!new JobConfig(config).getDiagnosticsEnabled()) {
      return;
    }
    // if diagnostics is enabled, create diagnostics stream if it doesnt exist

    String diagnosticsSystemStreamName = new MetricsConfig(config)
        .getMetricsSnapshotReporterStream(MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS)
        .orElseThrow(() -> new ConfigException("Missing required config: " +
            String.format(MetricsConfig.METRICS_SNAPSHOT_REPORTER_STREAM,
                MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS)));

    SystemStream diagnosticsSystemStream = StreamUtil.getSystemStreamFromNames(diagnosticsSystemStreamName);
    SystemConfig systemConfig = new SystemConfig(config);
    SystemAdmin diagnosticsSysAdmin = systemConfig.getSystemFactories().get(diagnosticsSystemStream.getSystem())
        .getAdmin(diagnosticsSystemStream.getSystem(), config, DiagnosticsUtil.class.getSimpleName());
    StreamSpec diagnosticsStreamSpec = new StreamSpec(DIAGNOSTICS_STREAM_ID, diagnosticsSystemStream.getStream(),
        diagnosticsSystemStream.getSystem(), new StreamConfig(config).getStreamProperties(DIAGNOSTICS_STREAM_ID));

    log.info("Creating diagnostics stream {}", diagnosticsSystemStream.getStream());
    diagnosticsSysAdmin.start();

    if (diagnosticsSysAdmin.createStream(diagnosticsStreamSpec)) {
      log.info("Created diagnostics stream {}", diagnosticsSystemStream.getStream());
    } else {
      log.info("Diagnostics stream {} already exists", diagnosticsSystemStream.getStream());
    }

    diagnosticsSysAdmin.stop();
  }
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