[kafka] branch trunk updated: KAFKA-14337; Correctly remove topicsWithCollisionChars after topic deletion (#12790)

[jgus]

This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new c1bb307a361 KAFKA-14337; Correctly remove topicsWithCollisionChars 
after topic deletion (#12790)
c1bb307a361 is described below

commit c1bb307a361b7b6e17261cc84ea2b108bacca84d
Author: Luke Chen <show...@gmail.com>
AuthorDate: Sat Oct 29 01:08:53 2022 +0800

    KAFKA-14337; Correctly remove topicsWithCollisionChars after topic deletion 
(#12790)
    
    In https://github.com/apache/kafka/pull/11910 , we added a feature to 
prevent topics with conflicting metrics names from being created. We added a 
map to store the normalized topic name to the topic names, but we didn't remove 
it correctly while deleting topics. This PR fixes this bug and add a test.
    
    Reviewers: Igor Soarez <i...@soarez.me>, dengziming 
<dengziming1...@gmail.com>, Jason Gustafson <ja...@confluent.io>
---
 .../kafka/admin/TopicCommandIntegrationTest.scala  | 25 ++++++++++++++++++++++
 .../controller/ReplicationControlManager.java      |  4 ++--
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git 
a/core/src/test/scala/integration/kafka/admin/TopicCommandIntegrationTest.scala 
b/core/src/test/scala/integration/kafka/admin/TopicCommandIntegrationTest.scala
index 3082babd06f..ea4c748da92 100644
--- 
a/core/src/test/scala/integration/kafka/admin/TopicCommandIntegrationTest.scala
+++ 
b/core/src/test/scala/integration/kafka/admin/TopicCommandIntegrationTest.scala
@@ -33,6 +33,7 @@ import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.MetadataResponse
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.function.Executable
 import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.ValueSource
@@ -470,6 +471,30 @@ class TopicCommandIntegrationTest extends 
KafkaServerTestHarness with Logging wi
     TestUtils.verifyTopicDeletion(zkClientOrNull, testTopicName, 1, brokers)
   }
 
+  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+  @ValueSource(strings = Array("zk", "kraft"))
+  def testTopicWithCollidingCharDeletionAndCreateAgain(quorum: String): Unit = 
{
+    // create the topic with colliding chars
+    val topicWithCollidingChar = "test.a"
+    val createOpts = new TopicCommandOptions(Array("--partitions", "1",
+      "--replication-factor", "1",
+      "--topic", topicWithCollidingChar))
+    createAndWaitTopic(createOpts)
+
+    // delete the topic
+    val deleteOpts = new TopicCommandOptions(Array("--topic", 
topicWithCollidingChar))
+
+    if (!isKRaftTest()) {
+      val deletePath = DeleteTopicsTopicZNode.path(topicWithCollidingChar)
+      assertFalse(zkClient.pathExists(deletePath), "Delete path for topic 
shouldn't exist before deletion.")
+    }
+    topicService.deleteTopic(deleteOpts)
+    TestUtils.verifyTopicDeletion(zkClientOrNull, topicWithCollidingChar, 1, 
brokers)
+
+    val createTopic: Executable = () => createAndWaitTopic(createOpts)
+    assertDoesNotThrow(createTopic)
+  }
+
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
   @ValueSource(strings = Array("zk", "kraft"))
   def testDeleteInternalTopic(quorum: String): Unit = {
diff --git 
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
 
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
index c48e1aed819..8c3e2880334 100644
--- 
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ 
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -342,7 +342,7 @@ public class ReplicationControlManager {
      * Since we reject topic creations that would collide, under normal 
conditions the
      * sets in this map should only have a size of 1. However, if the cluster 
was
      * upgraded from a version prior to KAFKA-13743, it may be possible to 
have more
-     * values here, since collidiing topic names will be "grandfathered in."
+     * values here, since colliding topic names will be "grandfathered in."
      */
     private final TimelineHashMap<String, TimelineHashSet<String>> 
topicsWithCollisionChars;
 
@@ -527,7 +527,7 @@ public class ReplicationControlManager {
             if (colliding != null) {
                 colliding.remove(topic.name);
                 if (colliding.isEmpty()) {
-                    topicsWithCollisionChars.remove(topic.name);
+                    topicsWithCollisionChars.remove(normalizedName);
                 }
             }
         }