private void populateCurrentLags(Set<SystemStreamPartition> ssps) {

    Map<MetricName, ? extends Metric> consumerMetrics = kafkaConsumer.metrics();

    // populate the MetricNames first time
    if (perPartitionMetrics.isEmpty()) {
      for (SystemStreamPartition ssp : ssps) {
        TopicPartition tp = KafkaSystemConsumer.toTopicPartition(ssp);

        // These are required by the KafkaConsumer to get the metrics
        HashMap<String, String> tags = new HashMap<>();
        tags.put("client-id", clientId);
        // kafka replaces '.' with underscore '_' in many/all of their metrics tags for topic names.
        // see https://github.com/apache/kafka/commit/5d81639907869ce7355c40d2bac176a655e52074#diff-b45245913eaae46aa847d2615d62cde0R1331
        tags.put("topic", tp.topic().replace('.', '_'));
        tags.put("partition", Integer.toString(tp.partition()));

        perPartitionMetrics.put(ssp, new MetricName("records-lag", "consumer-fetch-manager-metrics", "", tags));
      }
    }

    for (SystemStreamPartition ssp : ssps) {
      MetricName mn = perPartitionMetrics.get(ssp);
      Metric currentLagMetric = consumerMetrics.get(mn);

      // High watermark is fixed to be the offset of last available message,
      // so the lag is now at least 0, which is the same as Samza's definition.
      // If the lag is not 0, then isAtHead is not true, and kafkaClient keeps polling.
      long currentLag = (currentLagMetric != null) ? (long) currentLagMetric.value() : -1L;
      latestLags.put(ssp, currentLag);

      // calls the setIsAtHead for the BlockingEnvelopeMap
      sink.setIsAtHighWatermark(ssp, currentLag == 0);
    }
  }
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