[GitHub] [kafka] dhruvilshah3 commented on a change in pull request #10217: KAFKA-12254: Ensure MM2 creates topics with source topic configs

[GitBox <gi...@apache.org>]

dhruvilshah3 commented on a change in pull request #10217:
URL: https://github.com/apache/kafka/pull/10217#discussion_r583397577



##########
File path: connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java
##########
@@ -309,18 +309,37 @@ private void createOffsetSyncsTopic() {
     // visible for testing
     void computeAndCreateTopicPartitions()
             throws InterruptedException, ExecutionException {
-        Map<String, Long> partitionCounts = knownSourceTopicPartitions.stream()
+        Map<String, Long> topicToParitionCount = knownSourceTopicPartitions.stream()
             .collect(Collectors.groupingBy(TopicPartition::topic, Collectors.counting())).entrySet().stream()
             .collect(Collectors.toMap(x -> formatRemoteTopic(x.getKey()), Entry::getValue));
         Set<String> knownTargetTopics = toTopics(knownTargetTopicPartitions);
-        List<NewTopic> newTopics = partitionCounts.entrySet().stream()
-            .filter(x -> !knownTargetTopics.contains(x.getKey()))
-            .map(x -> new NewTopic(x.getKey(), x.getValue().intValue(), (short) replicationFactor))
+
+        // get the set of topics that are not present on the target
+        Set<String> topicsToCreate = topicToParitionCount.keySet().stream()
+                .filter(topic -> !knownTargetTopics.contains(topic))
+                .collect(Collectors.toSet());
+        if (topicsToCreate.isEmpty())
+            return;
+
+        // get corresponding topic configurations from the source
+        Map<String, Config> configs = describeTopicConfigs(topicsToCreate);

Review comment:
       This is not the right way to grab the topic configs as the names have been formatted using `formatRemoteTopic`.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org