private Runnable buildLoggingTask(String source, ReadableMetricsRegistry registry) {
    return () -> {
      for (String group : registry.getGroups()) {
        for (Map.Entry<String, Metric> metricGroupEntry : registry.getGroup(group).entrySet()) {
          metricGroupEntry.getValue().visit(new MetricsVisitor() {
            @Override
            public void counter(Counter counter) {
              logMetric(source, group, counter.getName(), counter.getCount());
            }

            @Override
            public <T> void gauge(Gauge<T> gauge) {
              logMetric(source, group, gauge.getName(), gauge.getValue());
            }

            @Override
            public void timer(Timer timer) {
              logMetric(source, group, timer.getName(), timer.getSnapshot().getAverage());
            }
          });
        }
      }
    };
  }
VisibleForTesting so that the logging call can be verified in unit tests.
samza-core/src/main/java/org/apache/samza/metrics/reporter/LoggingMetricsReporter.java