private OffsetsMaps fetchTopicPartitionsMetadata(List<TopicPartition> topicPartitions) {
    Map<SystemStreamPartition, String> oldestOffsets = new HashMap<>();
    Map<SystemStreamPartition, String> newestOffsets = new HashMap<>();
    Map<SystemStreamPartition, String> upcomingOffsets = new HashMap<>();
    final Map<TopicPartition, Long> oldestOffsetsWithLong = new HashMap<>();
    final Map<TopicPartition, Long> upcomingOffsetsWithLong = new HashMap<>();

    threadSafeKafkaConsumer.execute(consumer -> {
      Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(topicPartitions);
      LOG.debug("Beginning offsets for topic-partitions: {} is {}", topicPartitions, beginningOffsets);
      oldestOffsetsWithLong.putAll(beginningOffsets);
      Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);
      LOG.debug("End offsets for topic-partitions: {} is {}", topicPartitions, endOffsets);
      upcomingOffsetsWithLong.putAll(endOffsets);
      return Optional.empty();
    });

    oldestOffsetsWithLong.forEach((topicPartition, offset) -> oldestOffsets.put(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), String.valueOf(offset)));

    upcomingOffsetsWithLong.forEach((topicPartition, offset) -> {
      upcomingOffsets.put(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), String.valueOf(offset));

      // Kafka's beginning Offset corresponds to the offset for the oldest message.
      // Kafka's end offset corresponds to the offset for the upcoming message, and it is the newest offset + 1.
      // When upcoming offset is <=0, the topic appears empty, we put oldest offset 0 and the newest offset null.
      // When upcoming offset is >0, we subtract the upcoming offset by one for the newest offset.
      // For normal case, the newest offset will correspond to the offset of the newest message in the stream;
      // But for the big message, it is not the case. Seeking on the newest offset gives nothing for the newest big message.
      // For now, we keep it as is for newest offsets the same as historical metadata structure.
      if (offset <= 0) {
        LOG.warn(
            "Empty Kafka topic partition {} with upcoming offset {}. Skipping newest offset and setting oldest offset to 0 to consume from beginning",
            topicPartition, offset);
        oldestOffsets.put(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), "0");
      } else {
        newestOffsets.put(KafkaUtil.toSystemStreamPartition(systemName, topicPartition), String.valueOf(offset - 1));
      }
    });
    return new OffsetsMaps(oldestOffsets, newestOffsets, upcomingOffsets);
  }
Uses the kafka consumer to fetch the metadata for the {@code topicPartitions}.
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java