MINOR: Add test coverage for StorageTool format command feature validation (#20303)

### Summary
Adds comprehensive test coverage for the StorageTool format command
feature validation, including tests for valid feature overrides, invalid
feature detection, and multiple feature specifications. Also adds debug
output to help with troubleshooting format operations.

### Changes Made

#### Test Coverage Improvements
- **`testFormatWithReleaseVersionAndFeatureOverride()`**: Tests that
feature overrides work correctly when specified with `--feature` flag

- **`testFormatWithInvalidFeatureThrowsError()`**: Tests error handling
for unsupported features

- **`testFormatWithMultipleFeatures()`**: Tests multiple feature
specifications in a single format command

#### Debug Output Enhancement
- **Formatter.java**: Added debug output to print bootstrap metadata
during format operations
  - Helps with troubleshooting format issues by showing the complete
bootstrap metadata being written
  - Improves visibility into what features and configurations are being
applied

#### Test Updates
- **FormatterTest.java**: Updated existing tests to account for the new
debug output\

### Related
- KIP-1022: [Formatting and Updating Features

](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1022%3A+Formatting+and+Updating+Features)

Reviewers: Kevin Wu <[email protected]>, Justine Olshan
 <[email protected]>

Author: sjhajharia

core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

@@ -385,6 +385,84 @@ Found problem:
       "Failed to find content in output: " + stream.toString())
   }
 
+  @Test
+  def testFormatWithReleaseVersionAndFeatureOverride(): Unit = {
+    val availableDirs = Seq(TestUtils.tempDir())
+    val properties = new Properties()
+    properties.putAll(defaultStaticQuorumProperties)
+    properties.setProperty("log.dirs", availableDirs.mkString(","))
+    val stream = new ByteArrayOutputStream()
+    assertEquals(0, runFormatCommand(stream, properties, Seq(
+      "--release-version", "3.7-IV0",
+      "--feature", "share.version=1")))
+
+    // Verify that the feature override is applied by checking the bootstrap metadata
+    val bootstrapMetadata = new BootstrapDirectory(availableDirs.head.toString).read
+
+    // Verify that the share.version feature is set to 1 as specified
+    assertEquals(1.toShort, bootstrapMetadata.featureLevel("share.version"),
+      "share.version should be set to 1")
+
+    // Verify the command output contains the expected release version
+    assertTrue(stream.toString().contains("3.7-IV0"),
+      "Failed to find release version in output: " + stream.toString())
+
+    // Verify that the format command completed successfully with features
+    assertTrue(stream.toString().contains("Formatting metadata directory"),
+      "Failed to find formatting message in output: " + stream.toString())
+  }
+
+  @Test
+  def testFormatWithMultipleFeatures(): Unit = {
+    val availableDirs = Seq(TestUtils.tempDir())
+    val properties = new Properties()
+    properties.putAll(defaultStaticQuorumProperties)
+    properties.setProperty("log.dirs", availableDirs.mkString(","))
+    val stream = new ByteArrayOutputStream()
+    assertEquals(0, runFormatCommand(stream, properties, Seq(
+      "--release-version", "3.8-IV0",
+      "--feature", "share.version=1",
+      "--feature", "transaction.version=2",
+      "--feature", "group.version=1")))
+
+    // Verify that all features are properly bootstrapped by checking the bootstrap metadata
+    val bootstrapMetadata = new BootstrapDirectory(availableDirs.head.toString).read
+
+    // Verify that all specified features are set correctly
+    assertEquals(1.toShort, bootstrapMetadata.featureLevel("share.version"),
+      "share.version should be set to 1")
+    assertEquals(2.toShort, bootstrapMetadata.featureLevel("transaction.version"),
+      "transaction.version should be set to 2")
+    assertEquals(1.toShort, bootstrapMetadata.featureLevel("group.version"),
+      "group.version should be set to 1")
+
+    // Verify the command output contains the expected release version
+    assertTrue(stream.toString().contains("3.8-IV0"),
+      "Failed to find release version in output: " + stream.toString())
+
+    // Verify that the format command completed successfully with multiple features
+    assertTrue(stream.toString().contains("Formatting metadata directory"),
+      "Failed to find formatting message in output: " + stream.toString())
+  }
+  
+  @Test
+  def testFormatWithInvalidFeatureThrowsError(): Unit = {
+    val availableDirs = Seq(TestUtils.tempDir())
+    val properties = new Properties()
+    properties.putAll(defaultStaticQuorumProperties)
+    properties.setProperty("log.dirs", availableDirs.mkString(","))
+    val stream = new ByteArrayOutputStream()
+
+    // Test with an invalid feature that doesn't exist
+    val exception = assertThrows(classOf[FormatterException], () => {
+      runFormatCommand(stream, properties, Seq(
+        "--release-version", "3.7-IV0",
+        "--feature", "stream.version=1"))
+    })
+
+    assertTrue(exception.getMessage.contains("Unsupported feature: stream.version."))
+  }
+
   @Test
   def testFormatWithStandaloneFlagOnBrokerFails(): Unit = {
     val availableDirs = Seq(TestUtils.tempDir())

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

@@ -415,6 +415,7 @@ void doFormat(BootstrapMetadata bootstrapMetadata) throws Exception {
         if (ensemble.emptyLogDirs().isEmpty()) {
             printStream.println("All of the log directories are already formatted.");
         } else {
+            printStream.println("Bootstrap metadata: " + bootstrapMetadata);
             Map<String, DirectoryType> directoryTypes = new HashMap<>();
             for (String emptyLogDir : ensemble.emptyLogDirs()) {
                 DirectoryType directoryType = DirectoryType.calculate(emptyLogDir,

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

@@ -89,7 +89,7 @@ FormatterContext newFormatter() {
             Formatter formatter = new Formatter().
                 setNodeId(DEFAULT_NODE_ID).
                 setClusterId(DEFAULT_CLUSTER_ID.toString());
-            directories.forEach(d -> formatter.addDirectory(d));
+            directories.forEach(formatter::addDirectory);
             formatter.setMetadataLogDirectory(directories.get(0));
             return new FormatterContext(formatter);
         }
@@ -170,7 +170,7 @@ public void testFormatterFailsOnUnwritableDirectory() throws Exception {
             String expectedPrefix = "Error while writing meta.properties file";
             assertEquals(expectedPrefix,
                 assertThrows(FormatterException.class,
-                    () -> formatter1.formatter.run()).
+                    formatter1.formatter::run).
                         getMessage().substring(0, expectedPrefix.length()));
         }
     }
@@ -180,15 +180,15 @@ public void testIgnoreFormatted() throws Exception {
         try (TestEnv testEnv = new TestEnv(1)) {
             FormatterContext formatter1 = testEnv.newFormatter();
             formatter1.formatter.run();
-            assertEquals("Formatting metadata directory " + testEnv.directory(0) +
-                " with metadata.version " + MetadataVersion.latestProduction() + ".",
-                    formatter1.output().trim());
+            assertEquals("Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata() +
+                    "\nFormatting metadata directory " + testEnv.directory(0) +
+                    " with metadata.version " + MetadataVersion.latestProduction() + ".",
+                formatter1.output().trim());
 
             FormatterContext formatter2 = testEnv.newFormatter();
             formatter2.formatter.setIgnoreFormatted(true);
             formatter2.formatter.run();
-            assertEquals("All of the log directories are already formatted.",
-                formatter2.output().trim());
+            assertTrue(formatter2.output().trim().contains("All of the log directories are already formatted."));
         }
     }
 
@@ -201,7 +201,8 @@ public void testStandaloneWithIgnoreFormatted() throws Exception {
             formatter1.formatter
                 .setInitialControllers(DynamicVoters.parse("1@localhost:8020:" + originalDirectoryId))
                 .run();
-            assertEquals("Formatting dynamic metadata voter directory " + testEnv.directory(0) +
+            assertEquals("Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata() +
+                    "\nFormatting dynamic metadata voter directory " + testEnv.directory(0) +
                     " with metadata.version " + MetadataVersion.latestProduction() + ".",
                 formatter1.output().trim());
             assertMetadataDirectoryId(testEnv, Uuid.fromString(originalDirectoryId));
@@ -244,9 +245,10 @@ public void testOneDirectoryFormattedAndOthersNotFormattedWithIgnoreFormatted()
             FormatterContext formatter2 = testEnv.newFormatter();
             formatter2.formatter.setIgnoreFormatted(true);
             formatter2.formatter.run();
-            assertEquals("Formatting data directory " + testEnv.directory(1) + " with metadata.version " +
-                MetadataVersion.latestProduction() + ".",
-                    formatter2.output().trim());
+            assertEquals("Bootstrap metadata: " + formatter2.formatter.bootstrapMetadata() +
+                    "\nFormatting data directory " + testEnv.directory(1) + " with metadata.version " +
+                    MetadataVersion.latestProduction() + ".",
+                formatter2.output().trim());
         }
     }
 
@@ -256,9 +258,10 @@ public void testFormatWithOlderReleaseVersion() throws Exception {
             FormatterContext formatter1 = testEnv.newFormatter();
             formatter1.formatter.setReleaseVersion(MetadataVersion.IBP_3_5_IV0);
             formatter1.formatter.run();
-            assertEquals("Formatting metadata directory " + testEnv.directory(0) +
-                " with metadata.version " + MetadataVersion.IBP_3_5_IV0 + ".",
-                    formatter1.output().trim());
+            assertEquals("Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata() +
+                    "\nFormatting metadata directory " + testEnv.directory(0) +
+                    " with metadata.version " + MetadataVersion.IBP_3_5_IV0 + ".",
+                formatter1.output().trim());
             BootstrapMetadata bootstrapMetadata =
                 new BootstrapDirectory(testEnv.directory(0)).read();
             assertEquals(MetadataVersion.IBP_3_5_IV0, bootstrapMetadata.metadataVersion());
@@ -272,7 +275,7 @@ public void testFormatWithUnstableReleaseVersionFailsWithoutEnableUnstable() thr
             FormatterContext formatter1 = testEnv.newFormatter();
             formatter1.formatter.setReleaseVersion(MetadataVersion.latestTesting());
             assertEquals("metadata.version " + MetadataVersion.latestTesting() + " is not yet stable.",
-                assertThrows(FormatterException.class, () -> formatter1.formatter.run()).getMessage());
+                assertThrows(FormatterException.class, formatter1.formatter::run).getMessage());
         }
     }
 
@@ -283,9 +286,10 @@ public void testFormatWithUnstableReleaseVersion() throws Exception {
             formatter1.formatter.setReleaseVersion(MetadataVersion.latestTesting());
             formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
             formatter1.formatter.run();
-            assertEquals("Formatting metadata directory " + testEnv.directory(0) +
-                " with metadata.version " + MetadataVersion.latestTesting() + ".",
-                    formatter1.output().trim());
+            assertEquals("Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata() +
+                    "\nFormatting metadata directory " + testEnv.directory(0) +
+                    " with metadata.version " + MetadataVersion.latestTesting() + ".",
+                formatter1.output().trim());
             BootstrapMetadata bootstrapMetadata =
                     new BootstrapDirectory(testEnv.directory(0)).read();
             assertEquals(MetadataVersion.latestTesting(), bootstrapMetadata.metadataVersion());
@@ -318,7 +322,7 @@ public void testFormatWithScramFailsOnUnsupportedReleaseVersions() throws Except
                     "saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\"]"));
             assertEquals("SCRAM is only supported in metadata.version 3.5-IV2 or later.",
                 assertThrows(FormatterException.class,
-                    () -> formatter1.formatter.run()).getMessage());
+                    formatter1.formatter::run).getMessage());
         }
     }
 
@@ -333,9 +337,10 @@ public void testFormatWithScram() throws Exception {
                 "SCRAM-SHA-512=[name=alice,salt=\"MWx2NHBkbnc0ZndxN25vdGN4bTB5eTFrN3E=\"," +
                     "saltedpassword=\"mT0yyUUxnlJaC99HXgRTSYlbuqa4FSGtJCJfTMvjYCE=\"]"));
             formatter1.formatter.run();
-            assertEquals("Formatting metadata directory " + testEnv.directory(0) +
-                " with metadata.version " + MetadataVersion.IBP_3_8_IV0 + ".",
-                    formatter1.output().trim());
+            assertEquals("Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata() +
+                    "\nFormatting metadata directory " + testEnv.directory(0) +
+                    " with metadata.version " + MetadataVersion.IBP_3_8_IV0 + ".",
+                formatter1.output().trim());
             BootstrapMetadata bootstrapMetadata =
                 new BootstrapDirectory(testEnv.directory(0)).read();
             assertEquals(MetadataVersion.IBP_3_8_IV0, bootstrapMetadata.metadataVersion());
@@ -406,7 +411,7 @@ public void testInvalidFeatureFlag() throws Exception {
                     "are: eligible.leader.replicas.version, group.version, kraft.version, " +
                     "share.version, streams.version, test.feature.version, transaction.version",
                 assertThrows(FormatterException.class,
-                    () -> formatter1.formatter.run()).
+                    formatter1.formatter::run).
                         getMessage());
         }
     }
@@ -425,6 +430,7 @@ public void testFormatWithInitialVoters(boolean specifyKRaftVersion) throws Exce
             formatter1.formatter.run();
             assertEquals((short) 1, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 0));
             assertEquals(List.of(
+                "Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata(),
                 String.format("Formatting data directory %s with %s %s.",
                     testEnv.directory(1),
                     MetadataVersion.FEATURE_NAME,
@@ -459,7 +465,7 @@ public void testFormatWithInitialVotersFailsWithOlderKraftVersion() throws Excep
                 "Cannot set kraft.version to 0 if one of the flags --standalone, --initial-controllers, or " +
                 "--no-initial-controllers is used. For dynamic controllers support, try removing the " +
                 "--feature flag for kraft.version.",
-                assertThrows(FormatterException.class, () -> formatter1.formatter.run()).getMessage()
+                assertThrows(FormatterException.class, formatter1.formatter::run).getMessage()
             );
         }
     }
@@ -475,7 +481,7 @@ public void testFormatWithoutInitialVotersFailsWithNewerKraftVersion() throws Ex
                 "Cannot set kraft.version to 1 unless one of the flags --standalone, --initial-controllers, or " +
                 "--no-initial-controllers is used. For dynamic controllers support, try using one of " +
                 "--standalone, --initial-controllers, or --no-initial-controllers.",
-                assertThrows(FormatterException.class, () -> formatter1.formatter.run()).getMessage()
+                assertThrows(FormatterException.class, formatter1.formatter::run).getMessage()
             );
         }
     }
@@ -492,7 +498,7 @@ public void testFormatWithInitialVotersFailsWithOlderMetadataVersion() throws Ex
             assertEquals("kraft.version could not be set to 1 because it depends on " +
                 "metadata.version level 21",
                     assertThrows(IllegalArgumentException.class,
-                        () -> formatter1.formatter.run()).getMessage());
+                        formatter1.formatter::run).getMessage());
         }
     }
 
@@ -514,12 +520,12 @@ public void testFormatElrEnabledWithMetadataVersions(MetadataVersion metadataVer
             formatter1.formatter.setInitialControllers(DynamicVoters.
                 parse("1@localhost:8020:4znU-ou9Taa06bmEJxsjnw"));
             if (metadataVersion.isAtLeast(MetadataVersion.IBP_4_0_IV1)) {
-                assertDoesNotThrow(() -> formatter1.formatter.run());
+                assertDoesNotThrow(formatter1.formatter::run);
             } else {
                 assertEquals("eligible.leader.replicas.version could not be set to 1 because it depends on " +
                     "metadata.version level 23",
                     assertThrows(IllegalArgumentException.class,
-                        () -> formatter1.formatter.run()).getMessage());
+                        formatter1.formatter::run).getMessage());
             }
         }
     }
@@ -539,6 +545,7 @@ public void testFormatWithNoInitialControllers(boolean specifyKRaftVersion) thro
             formatter1.formatter.run();
             assertEquals((short) 1, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 0));
             assertEquals(List.of(
+                    "Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata(),
                     String.format("Formatting data directory %s with %s %s.",
                         testEnv.directory(1),
                         MetadataVersion.FEATURE_NAME,