private Map<String, SystemStreamMetadata> fetchSystemStreamMetadata(Set<String> topics) {
    Map<SystemStreamPartition, String> allOldestOffsets = new HashMap<>();
    Map<SystemStreamPartition, String> allNewestOffsets = new HashMap<>();
    Map<SystemStreamPartition, String> allUpcomingOffsets = new HashMap<>();

    LOG.info("Fetching SystemStreamMetadata for topics {} on system {}", topics, systemName);

    topics.forEach(topic -> {
      OffsetsMaps offsetsForTopic = threadSafeKafkaConsumer.execute(consumer -> {
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        if (partitionInfos == null) {
          String msg = String.format("Partition info not(yet?) available for system %s topic %s", systemName, topic);
          throw new SamzaException(msg);
        }
        List<TopicPartition> topicPartitions = partitionInfos.stream()
            .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))
            .collect(Collectors.toList());
        return fetchTopicPartitionsMetadata(topicPartitions);
      });
      allOldestOffsets.putAll(offsetsForTopic.getOldestOffsets());
      allNewestOffsets.putAll(offsetsForTopic.getNewestOffsets());
      allUpcomingOffsets.putAll(offsetsForTopic.getUpcomingOffsets());
    });

    return assembleMetadata(allOldestOffsets, allNewestOffsets, allUpcomingOffsets);
  }
Fetch SystemStreamMetadata for each topic with the consumer
@param topics set of topics to get metadata info for
@return map of topic to SystemStreamMetadata
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java