Address review comments

Author: gaurav-narula

metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java

@@ -229,27 +229,36 @@ public BootstrapMetadata bootstrapMetadata() {
         return bootstrapMetadata;
     }
 
-    public void run() throws Exception {
-        if (nodeId < 0) {
-            throw new RuntimeException("You must specify a valid non-negative node ID.");
-        }
+    /**
+     * Validates the correctness of the given cluster id. A valid cluster id is a base64, urlencoded, no padding
+     * representation of a {@link Uuid}. These checks do not validate the absence of <code>-</code> character as
+     * {@link Uuid#randomUuid()} avoids them only for convenience reasons.
+     */
+    private void validateClusterId(String clusterId) {
         if (clusterId == null) {
             throw new FormatterException("You must specify the cluster id.");
         }
+        if (clusterId.contains("=")) {
+            throw new FormatterException("The specified cluster id, " + clusterId + " is invalid: contains padding");
+        }
         try {
-            if (clusterId.contains("=")) {
-                throw new FormatterException("The specified cluster id, " + clusterId + " contains padding and is invalid");
-            }
             Uuid uuid = Uuid.fromString(clusterId);
             if (Uuid.RESERVED.contains(uuid)) {
                 throw new FormatterException("The specified cluster id, " + clusterId + " is reserved");
             }
         } catch (IllegalArgumentException e) {
             throw new FormatterException("The specified cluster id, " + clusterId + " is invalid", e);
         }
+    }
+
+    public void run() throws Exception {
+        if (nodeId < 0) {
+            throw new RuntimeException("You must specify a valid non-negative node ID.");
+        }
         if (directories.isEmpty()) {
             throw new FormatterException("You must specify at least one directory to format");
         }
+        validateClusterId(clusterId);
         if (controllerListenerName == null) {
             throw new FormatterException("You must specify the name of the initial controller listener.");
         }

metadata/src/test/java/org/apache/kafka/metadata/storage/FormatterTest.java

@@ -601,4 +601,14 @@ public void testFormatWithInvalidClusterId(String clusterId) throws Exception {
                             getMessage().substring(0, expectedPrefix.length()));
         }
     }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"iIygKoNNSpGzNeAEr_QW7w", "-w_KF-snTmuEnKPqZ0RDzA", "dZMgZA7nTLqZTdzb0zbvSQ", "hxqHyN2OSSmPajVhm_DR-Q"})
+    public void testFormatWithValidClusterId(String clusterId) throws Exception {
+        try (TestEnv testEnv = new TestEnv(2)) {
+            FormatterContext formatter1 = testEnv.newFormatter();
+            formatter1.formatter.setClusterId(clusterId);
+            formatter1.formatter.run();
+        }
+    }
 }