filter no leader

TongchengOpenSource · Dec 17, 2024 · d188f6e · d188f6e
1 parent a139595
commit d188f6e
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...va/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java b/...va/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java
@@ -307,10 +307,13 @@ private Set<KafkaSourceSplit> getTopicInfo() throws ExecutionException, Interrup
         }
         log.info("Discovered topics: {}", topics);
         Collection<TopicPartition> partitions =
-                adminClient.describeTopics(topics).all().get().values().stream()
+                adminClient.describeTopics(topics).allTopicNames().get().values().stream()
                         .flatMap(
                                 t ->
                                         t.partitions().stream()
+                                                .filter(
+                                                        partitionInfo ->
+                                                                partitionInfo.leader() != null)
                                                 .map(
                                                         p ->
                                                                 new TopicPartition(