KAFKA-20617: add validation for cluster-id in Formatter

This change adds validation for Uuids in Formatter which is used by
`kafka-storage.sh` which allows users to fail fast before they
inadvertently end up using an invalid cluster-id.

Author: gaurav-narula

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

@@ -236,6 +236,17 @@ public void run() throws Exception {
         if (clusterId == null) {
             throw new FormatterException("You must specify the cluster id.");
         }
+        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);
+        }
         if (directories.isEmpty()) {
             throw new FormatterException("You must specify at least one directory to format");
         }

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

@@ -587,4 +587,18 @@ public void testFormatWithNoInitialControllers() throws Exception {
             assertNotNull(logDirProps1);
         }
     }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"unrvTtQISjar0JUWGU/8Pg", "igNUVIdeSPO5JCZYFhOh7Q==", "AAAAAAAAAAAAAAAAAAAAAA", "AAAAAAAAAAAAAAAAAAAAAQ"})
+    public void testFormatWithInvalidClusterId(String clusterId) throws Exception {
+        try (TestEnv testEnv = new TestEnv(2)) {
+            FormatterContext formatter1 = testEnv.newFormatter();
+            formatter1.formatter.setClusterId(clusterId);
+            String expectedPrefix = "The specified cluster id, " + clusterId;
+            assertEquals(expectedPrefix,
+                    assertThrows(FormatterException.class,
+                            formatter1.formatter::run).
+                            getMessage().substring(0, expectedPrefix.length()));
+        }
+    }
 }