public void updateMetrics(String stream, Record record) {
    eventReadRates.get(stream).inc();
    aggEventReadRate.inc();

    long recordSize = record.getData().array().length + record.getPartitionKey().length();
    eventByteReadRates.get(stream).inc(recordSize);
    aggEventByteReadRate.inc(recordSize);

    long latencyMs = Duration.between(Instant.now(), record.getApproximateArrivalTimestamp().toInstant()).toMillis();
    readLatencies.get(stream).update(latencyMs);
    aggReadLatency.update(latencyMs);
  }
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