KAFKA-20072 Make Uuid no-dash behavior opt-in

clients/src/main/java/org/apache/kafka/common/Uuid.java

@@ -71,9 +71,24 @@ private static Uuid unsafeRandomUuid() {
     /**
      * Static factory to retrieve a type 4 (pseudo randomly generated) UUID.
      * <p>
-     * This will not generate a UUID equal to 0, 1, or one whose string representation contains a dash ("-").
+     * This will not generate a UUID equal to 0, 1, or one whose string representation starts with a dash ("-").
      */
     public static Uuid randomUuid() {
+        Uuid uuid = unsafeRandomUuid();
+        while (RESERVED.contains(uuid) || uuid.toString().startsWith("-")) {
+            uuid = unsafeRandomUuid();
+        }
+        return uuid;
+    }
+
+    /**
+     * Static factory to retrieve a type 4 (pseudo randomly generated) UUID.
+     * <p>
+     * This will not generate a UUID equal to 0, 1, or one whose string representation contains a dash ("-").
+     * Note that this will result in 30% rejected UUIDs, and so should not be used in performance sensitive
+     * code paths.
+     */
+    public static Uuid randomUuidNoDashes() {
         Uuid uuid = unsafeRandomUuid();
         while (RESERVED.contains(uuid) || uuid.toString().contains("-")) {
             uuid = unsafeRandomUuid();

clients/src/test/java/org/apache/kafka/common/UuidTest.java

@@ -81,6 +81,15 @@ public void testStringConversion() {
     public void testRandomUuid() {
         Uuid randomID = Uuid.randomUuid();
 
+        assertNotEquals(Uuid.ZERO_UUID, randomID);
+        assertNotEquals(Uuid.METADATA_TOPIC_ID, randomID);
+        assertFalse(randomID.toString().startsWith("-"));
+    }
+
+    @RepeatedTest(value = 100, name = RepeatedTest.LONG_DISPLAY_NAME)
+    public void testRandomUuidNoDash() {
+        Uuid randomID = Uuid.randomUuidNoDashes();
+
         assertNotEquals(Uuid.ZERO_UUID, randomID);
         assertNotEquals(Uuid.METADATA_TOPIC_ID, randomID);
         assertFalse(randomID.toString().contains("-"));

docs/design/protocol.md

@@ -218,11 +218,11 @@ A final question is why we don't use a system like Protocol Buffers or Thrift to
 
 ## Recommendations for 3rd‑party Clients: Member ID Format
 
-When a Kafka client participates in group protocols (e.g., `ConsumerGroupHeartbeat` RPC), it must generate a **member ID** to identify itself to the broker. While the protocol does not strictly enforce the format of this ID, we strongly recommend the following:
+When a Kafka client participates in group protocols (e.g., `ConsumerGroupHeartbeat` RPC), it must generate a **member ID** to identify itself to the broker. While the protocol does not strictly enforce the format of this ID, we recommend the following:
 
 1. **Use a base64‑encoded UUID** as the member ID.
 2. **Encode the UUID using URL‑safe base64** (without `+` or `/` characters).
-3. **Omit hyphens** — the resulting string should be a continuous sequence of alphanumeric characters (e.g., `abc123def456`).
+3. **Omit leading hyphens** — the resulting string should not include a leading hyphen
 
 **Example**  
 A standard UUID (`00000000-0000-0000-0000-000000000000`) should be transformed into a URL‑safe base64 string like: `YzYxNjQ4OTItZDE1Mi00Y2E4LWIyNzUtYmIwMzAwMDAwMDAw`

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -815,7 +815,7 @@ private ApiError createTopic(ControllerRequestContext context,
                 numPartitions, e.throttleTimeMs());
             return ApiError.fromThrowable(e);
         }
-        Uuid topicId = Uuid.randomUuid();
+        Uuid topicId = Uuid.randomUuidNoDashes();
         CreatableTopicResult result = new CreatableTopicResult().
             setName(topic.name()).
             setTopicId(topicId).