public void innerRun() {
    LOG.debug("Begin flushing metrics.");
    for (MetricsRegistryWithSource metricsRegistryWithSource : this.registries) {
      String source = metricsRegistryWithSource.getSource();
      ReadableMetricsRegistry registry = metricsRegistryWithSource.getRegistry();
      LOG.debug("Flushing metrics for {}.", source);
      Map<String, Map<String, Object>> metricsMsg = new HashMap<>();

      // metrics
      registry.getGroups().forEach(group -> {
        Map<String, Object> groupMsg = new HashMap<>();
        registry.getGroup(group).forEach((name, metric) -> {
          if (!shouldIgnore(group, name)) {
            metric.visit(new MetricsVisitor() {
              @Override
              public void counter(Counter counter) {
                groupMsg.put(name, counter.getCount());
              }

              @Override
              public <T> void gauge(Gauge<T> gauge) {
                groupMsg.put(name, gauge.getValue());
              }

              @Override
              public void timer(Timer timer) {
                groupMsg.put(name, timer.getSnapshot().getAverage());
              }
            });
          }
        });

        // dont emit empty groups
        if (!groupMsg.isEmpty()) {
          metricsMsg.put(group, groupMsg);
        }
      });

      // publish to Kafka only if the metricsMsg carries any metrics
      if (!metricsMsg.isEmpty()) {
        MetricsHeader header =
            new MetricsHeader(this.jobName, this.jobId, this.containerName, this.executionEnvContainerId,
                Optional.of(this.samzaEpochId), source, this.version, this.samzaVersion, this.host,
                this.clock.currentTimeMillis(), this.resetTime);
        Metrics metrics = new Metrics(metricsMsg);
        LOG.debug("Flushing metrics for {} to {} with header and map: header={}, map={}.", source, out,
            header.getAsMap(), metrics.getAsMap());
        MetricsSnapshot metricsSnapshot = new MetricsSnapshot(header, metrics);
        Object maybeSerialized = (this.serializer != null) ? this.serializer.toBytes(metricsSnapshot) : metricsSnapshot;
        try {
          this.producer.send(source, new OutgoingMessageEnvelope(this.out, this.host, null, maybeSerialized));
          // Always flush, since we don't want metrics to get batched up.
          this.producer.flush(source);
        } catch (Exception e) {
          LOG.error(String.format("Exception when flushing metrics for source %s", source), e);
        }
      }
    }
    LOG.debug("Finished flushing metrics.");
  }
MetricsSnapshotReporter is a generic metrics reporter that sends metrics to a stream.

jobName // my-samza-job
jobId // an id that differentiates multiple executions of the same job
taskName // container_567890
host // eat1-app128.gird
version // 0.0.1
blacklist // Regex of metrics to ignore when flushing
samza-core/src/main/java/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.java