MINOR: Add test case for topic recreation with collision chars (apach…

…e#12796) This patch adds a unit test for topic recreation with colliding characters (such as `.`). This was broken up until apache#12790. Reviewers: José Armando García Sancio <[email protected]>
EricMinick · Oct 29, 2022 · cbe50d9 · cbe50d9
1 parent c1bb307
commit cbe50d9
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 0 deletions.
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -263,6 +263,10 @@ public String name() {
         public Uuid topicId() {
             return id;
         }
+
+        public int numPartitions(long epoch) {
+            return parts.size(epoch);
+        }
     }
 
     /**

diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -60,6 +60,7 @@
 import org.apache.kafka.common.metadata.PartitionChangeRecord;
 import org.apache.kafka.common.metadata.PartitionRecord;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord;
+import org.apache.kafka.common.metadata.RemoveTopicRecord;
 import org.apache.kafka.common.metadata.TopicRecord;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
@@ -269,6 +270,21 @@ CreatableTopicResult createTestTopic(String name, int[][] replicas,
             return topicResult;
         }
 
+        void deleteTopic(Uuid topicId) throws Exception {
+            ControllerResult<Map<Uuid, ApiError>> result = replicationControl.deleteTopics(Collections.singleton(topicId));
+            assertEquals(Collections.singleton(topicId), result.response().keySet());
+            assertEquals(NONE, result.response().get(topicId).error());
+            assertEquals(1, result.records().size());
+
+            ApiMessageAndVersion removeRecordAndVersion = result.records().get(0);
+            assertTrue(removeRecordAndVersion.message() instanceof RemoveTopicRecord);
+
+            RemoveTopicRecord removeRecord = (RemoveTopicRecord) removeRecordAndVersion.message();
+            assertEquals(topicId, removeRecord.topicId());
+
+            replay(result.records());
+        }
+
         void createPartitions(int count, String name,
                 int[][] replicas, short expectedErrorCode) throws Exception {
             assertFalse(replicas.length == 0);
@@ -698,6 +714,21 @@ public void testCreateTopicsWithPolicy() throws Exception {
         ctx.createTestTopic("quux", new int[][] {new int[] {1, 2, 0}}, POLICY_VIOLATION.code());
     }
 
+    @Test
+    public void testCreateTopicWithCollisionChars() throws Exception {
+        ReplicationControlTestContext ctx = new ReplicationControlTestContext(Optional.empty());
+        ctx.registerBrokers(0, 1, 2);
+        ctx.unfenceBrokers(0, 1, 2);
+
+        CreatableTopicResult initialTopic = ctx.createTestTopic("foo.bar", 2, (short) 2, NONE.code());
+        assertEquals(2, ctx.replicationControl.getTopic(initialTopic.topicId()).numPartitions(Long.MAX_VALUE));
+        ctx.deleteTopic(initialTopic.topicId());
+
+        CreatableTopicResult recreatedTopic = ctx.createTestTopic("foo.bar", 4, (short) 2, NONE.code());
+        assertNotEquals(initialTopic.topicId(), recreatedTopic.topicId());
+        assertEquals(4, ctx.replicationControl.getTopic(recreatedTopic.topicId()).numPartitions(Long.MAX_VALUE));
+    }
+
     @Test
     public void testGlobalTopicAndPartitionMetrics() throws Exception {
         ReplicationControlTestContext ctx = new ReplicationControlTestContext();