[FEATURE] Add getTopics API to StreamMetadataProvider for Topic Enumeration in Streaming Connectors

pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.common.PartitionInfo;
@@ -166,6 +167,18 @@ public Map<String, PartitionLagState> getCurrentPartitionLagState(
     return perPartitionLag;
   }
 
+  @Override
+  public List<TopicMetadata> listTopics(Duration timeout) {
+    Map<String, List<PartitionInfo>> namePartitionsMap = _consumer.listTopics(timeout);
+    if (namePartitionsMap == null) {
+      return Collections.emptyList();
+    }
+    return namePartitionsMap.keySet()
+        .stream()
+        .map(topic -> new TopicMetadata().setName(topic))
+        .collect(Collectors.toList());
+  }
+
   @Override
   public void close()
       throws IOException {

pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.common.PartitionInfo;
@@ -166,6 +167,18 @@ public Map<String, PartitionLagState> getCurrentPartitionLagState(
     return perPartitionLag;
   }
 
+  @Override
+  public List<TopicMetadata> listTopics(Duration timeout) {
+    Map<String, List<PartitionInfo>> namePartitionsMap = _consumer.listTopics(timeout);
+    if (namePartitionsMap == null) {
+      return Collections.emptyList();
+    }
+    return namePartitionsMap.keySet()
+        .stream()
+        .map(topic -> new TopicMetadata().setName(topic))
+        .collect(Collectors.toList());
+  }
+
   @Override
   public void close()
       throws IOException {

pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java

@@ -18,11 +18,15 @@
  */
 package org.apache.pinot.plugin.stream.kafka30;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -39,6 +43,7 @@
 import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
 import org.apache.pinot.spi.stream.StreamMessage;
 import org.apache.pinot.spi.stream.StreamMessageMetadata;
+import org.apache.pinot.spi.stream.StreamMetadataProvider;
 import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -399,4 +404,29 @@ public void testOffsetsExpired()
     }
     assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "700");
   }
+
+  @Test
+  public void testListTopics() {
+    String streamType = "kafka";
+    String streamKafkaBrokerList = _kafkaBrokerAddress;
+    String streamKafkaConsumerType = "simple";
+    String clientId = "clientId";
+    String tableNameWithType = "tableName_REALTIME";
+
+    Map<String, String> streamConfigMap = new HashMap<>();
+    streamConfigMap.put("streamType", streamType);
+    streamConfigMap.put("stream.kafka.topic.name", "NON_EXISTING_TOPIC");
+    streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+    streamConfigMap.put("stream.kafka.consumer.type", streamKafkaConsumerType);
+    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+    StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+    KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
+    List<StreamMetadataProvider.TopicMetadata> topics = streamMetadataProvider.listTopics(Duration.ofSeconds(60));
+    Set<String> topicNames = topics.stream()
+        .map(StreamMetadataProvider.TopicMetadata::getName)
+        .collect(Collectors.toSet());
+    assertEquals(topicNames, Set.of(TEST_TOPIC_1, TEST_TOPIC_2, TEST_TOPIC_3));
+  }
 }

pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java

@@ -20,6 +20,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -106,6 +107,33 @@ default Map<String, PartitionLagState> getCurrentPartitionLagState(
     return result;
   }
 
+  /**
+   * Fetches the list of available topics/streams
+   *
+   * @param timeout Timeout for fetching the list of topics. If this is null, the implementation should use a default
+   *                timeout.
+   * @return List of topics
+   */
+  default List<TopicMetadata> listTopics(Duration timeout) {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Represents the metadata of a topic. This can be used to represent the topic name and other metadata in the future.
+   */
+  class TopicMetadata {
+    private String _name;
+
+    public String getName() {
+      return _name;
+    }
+
+    public TopicMetadata setName(String name) {
+      _name = name;
+      return this;
+    }
+  }
+
   class UnknownLagState extends PartitionLagState {
   }
 }