private void updateMetrics(ConsumerRecord<K, V> r, TopicPartition tp) {
    SystemStreamPartition ssp = new SystemStreamPartition(systemName, tp.topic(), new Partition(tp.partition()));

    Long lag = latestLags.get(ssp);
    if (lag == null) {
      throw new SamzaException("Unknown/unregistered ssp in latestLags. ssp=" + ssp + "; system=" + systemName);
    }
    long currentSSPLag = lag; // lag between the current offset and the highwatermark
    if (currentSSPLag < 0) {
      return;
    }

    long recordOffset = r.offset();
    long highWatermark = recordOffset + currentSSPLag; // derived value for the highwatermark

    int size = getRecordSize(r);
    kafkaConsumerMetrics.incReads(tp);
    kafkaConsumerMetrics.incBytesReads(tp, size);
    kafkaConsumerMetrics.setOffsets(tp, recordOffset);
    kafkaConsumerMetrics.incClientBytesReads(metricName, size);
    kafkaConsumerMetrics.setHighWatermarkValue(tp, highWatermark);
  }
Protected to help extensions of this class build {@link IncomingMessageEnvelope}s.
@param r consumer record to size
@return the size of the serialized record
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java