private void refreshLagMetrics() {
    for (Map.Entry<SystemStreamPartition, Long> e : nextOffsets.entrySet()) {
      SystemStreamPartition ssp = e.getKey();
      Long offset = e.getValue();
      TopicPartition tp = new TopicPartition(ssp.getStream(), ssp.getPartition().getPartitionId());
      Long lag = latestLags.get(ssp);
      LOG.trace("Latest offset of {} is  {}; lag = {}", ssp, offset, lag);
      if (lag != null && offset != null && lag >= 0) {
        long streamEndOffset = offset + lag;
        // update the metrics
        kafkaConsumerMetrics.setHighWatermarkValue(tp, streamEndOffset);
        kafkaConsumerMetrics.setLagValue(tp, lag);
      }
    }
  }
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