public void updateMillisBehindLatest(String stream, Long millisBehindLatest) {
    this.millisBehindLatest.get(stream).update(millisBehindLatest);
    aggMillisBehindLatest.update(millisBehindLatest);
  }
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