KAFKA-13743: Prevent topics with conflicting metrics names from being…

… created in KRaft mode apache#11910

In ZK mode, the topic "foo_bar" will conflict with "foo.bar" because of limitations in metric
names. We should implement this in KRaft mode.  This PR also changes TopicCommandIntegrationTest to
support KRaft mode.

Reviewers: Colin P. McCabe <cmccabe@apache.org>

clients/src/main/java/org/apache/kafka/common/internals/Topic.java

@@ -67,6 +67,17 @@ public static boolean hasCollisionChars(String topic) {
         return topic.contains("_") || topic.contains(".");
     }
 
+    /**
+     * Unify topic name with a period ('.') or underscore ('_'), this is only used to check collision and will not
+     * be used to really change topic name.
+     *
+     * @param topic A topic to unify
+     * @return A unified topic name
+     */
+    public static String unifyCollisionChars(String topic) {
+        return topic.replace('.', '_');
+    }
+
     /**
      * Returns true if the topicNames collide due to a period ('.') or underscore ('_') in the same position.
      *
@@ -75,7 +86,7 @@ public static boolean hasCollisionChars(String topic) {
      * @return true if the topics collide
      */
     public static boolean hasCollision(String topicA, String topicB) {
-        return topicA.replace('.', '_').equals(topicB.replace('.', '_'));
+        return unifyCollisionChars(topicA).equals(unifyCollisionChars(topicB));
     }
 
     /**

clients/src/test/java/org/apache/kafka/common/internals/TopicTest.java

@@ -24,6 +24,7 @@
 import java.util.Collections;
 import java.util.List;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -81,6 +82,14 @@ public void testTopicHasCollisionChars() {
             assertTrue(Topic.hasCollisionChars(topic));
     }
 
+    @Test
+    public void testUnifyCollisionChars() {
+        assertEquals("topic", Topic.unifyCollisionChars("topic"));
+        assertEquals("_topic", Topic.unifyCollisionChars(".topic"));
+        assertEquals("_topic", Topic.unifyCollisionChars("_topic"));
+        assertEquals("__topic", Topic.unifyCollisionChars("_.topic"));
+    }
+
     @Test
     public void testTopicHasCollision() {
         List<String> periodFirstMiddleLastNone = Arrays.asList(".topic", "to.pic", "topic.", "topic");