[branch-2.7.2] Fix unused getList call to create missed partitions (#626

) Fixes #619 See #619 (comment) for details.
streamnative · Jul 26, 2021 · 5da3a66 · 5da3a66
1 parent 96f454a
commit 5da3a66
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 14 deletions.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtils.java
@@ -22,7 +22,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -150,18 +149,10 @@ public static void createKafkaMetadataIfMissing(PulsarAdmin pulsarAdmin, KafkaSe
                 log.info("Successfully created group metadata topic {} with {} partitions.",
                         offsetsTopic, conf.getOffsetsTopicNumPartitions());
             } else {
-                // Check to see if the partitions all exist
-                offsetPartitionSet.removeAll(
-                topics.getList(kafkaMetadataNamespace).stream()
-                .filter((topic) -> {
-                    return topic.startsWith(offsetsTopic + PARTITIONED_TOPIC_SUFFIX);
-                }).collect(Collectors.toList()));
-
-                if (!offsetPartitionSet.isEmpty()) {
-                    log.info("Identified missing offset topic partitions: {}", offsetPartitionSet);
-                    for (String offsetPartition : offsetPartitionSet) {
-                        topics.createNonPartitionedTopic(offsetPartition);
-                    }
+                try {
+                    // Ensure all partitions are created
+                    pulsarAdmin.topics().createMissedPartitions(offsetsTopic);
+                } catch (PulsarAdminException ignored) {
                 }
             }
             offsetsTopicExists = true;

diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtilsTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtilsTest.java
@@ -123,6 +123,6 @@ public void testCreateKafkaMetadataIfMissing() throws Exception {
         verify(mockTenants, times(1)).updateTenant(eq(conf.getKafkaMetadataTenant()), any(TenantInfo.class));
         verify(mockNamespaces, times(1)).setNamespaceReplicationClusters(eq(conf.getKafkaMetadataTenant()
             + "/" + conf.getKafkaMetadataNamespace()), any(Set.class));
-        verify(mockTopics, times(2)).createNonPartitionedTopic(any(String.class));
+        verify(mockTopics, times(1)).createMissedPartitions(any(String.class));
     }
 }