private void initializeLags() {
    // This is expensive, so only do it once at the beginning. After the first poll, we can rely on metrics for lag.

    Map<TopicPartition, Long> endOffsets;
    // Synchronize, in case the consumer is used in some other thread (metadata or something else)
    synchronized (kafkaConsumer) {
      endOffsets = kafkaConsumer.endOffsets(topicPartitionToSSP.keySet());
    }
    if (endOffsets == null) {
      throw new SamzaException("Failed to fetch kafka consumer endoffsets for system " + systemName);
    }
    endOffsets.forEach((tp, offset) -> {
      SystemStreamPartition ssp = topicPartitionToSSP.get(tp);
      long startingOffset = nextOffsets.get(ssp);
      // End offsets are the offset of the newest message + 1
      // If the message we are about to consume is < end offset, we are starting with a lag.
      long initialLag = endOffsets.get(tp) - startingOffset;

      LOG.info("Initial lag for SSP {} is {} (end={}, startOffset={})", ssp, initialLag, endOffsets.get(tp), startingOffset);
      latestLags.put(ssp, initialLag);
      sink.setIsAtHighWatermark(ssp, initialLag == 0);
    });

    // initialize lag metrics
    refreshLagMetrics();
  }
Stop this KafkaConsumerProxy and wait for at most {@code timeoutMs}.
@param timeoutMs maximum time to wait to stop this KafkaConsumerProxy
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java