public void initializeMetrics(Set<String> streamNames) {
    eventReadRates = streamNames.stream()
        .collect(Collectors.toConcurrentMap(Function.identity(), x -> registry.newCounter(x, EVENT_READ_RATE)));
    eventByteReadRates = streamNames.stream()
        .collect(Collectors.toConcurrentMap(Function.identity(), x -> registry.newCounter(x, EVENT_BYTE_READ_RATE)));
    readLatencies = streamNames.stream()
        .collect(Collectors.toConcurrentMap(Function.identity(), x -> new SamzaHistogram(registry, x, READ_LATENCY)));
    millisBehindLatest = streamNames.stream()
        .collect(Collectors.toConcurrentMap(Function.identity(),
          x -> new SamzaHistogram(registry, x, MILLIS_BEHIND_LATEST)));

    // Locking to ensure that these aggregated metrics will be created only once across multiple system consumers.
    synchronized (LOCK) {
      if (aggEventReadRate == null) {
        aggEventReadRate = registry.newCounter(AGGREGATE, EVENT_READ_RATE);
        aggEventByteReadRate = registry.newCounter(AGGREGATE, EVENT_BYTE_READ_RATE);
        aggReadLatency = new SamzaHistogram(registry, AGGREGATE, READ_LATENCY);
        aggMillisBehindLatest = new SamzaHistogram(registry, AGGREGATE, MILLIS_BEHIND_LATEST);
      }
    }
  }
KinesisSystemConsumerMetrics class has per-stream metrics and aggregate metrics across kinesis consumers
samza-aws/src/main/java/org/apache/samza/system/kinesis/metrics/KinesisSystemConsumerMetrics.java