KAFKA-19719: --no-initial-controllers should not assume kraft.version=1

core/src/main/scala/kafka/tools/StorageTool.scala

@@ -163,16 +163,13 @@ object StorageTool extends Logging {
     if (isStandalone) {
       formatter.setInitialControllers(createStandaloneDynamicVoters(config))
     }
-    if (namespace.getBoolean("no_initial_controllers")) {
-      formatter.setNoInitialControllersFlag(true)
-    } else {
-      if (config.processRoles.contains(ProcessRole.ControllerRole)) {
-        if (config.quorumConfig.voters().isEmpty && formatter.initialVoters().isEmpty) {
+    if (!namespace.getBoolean("no_initial_controllers") &&
+      config.processRoles.contains(ProcessRole.ControllerRole) &&
+      config.quorumConfig.voters().isEmpty &&
+      formatter.initialVoters().isEmpty) {
           throw new TerseFailure("Because " + QuorumConfig.QUORUM_VOTERS_CONFIG +
             " is not set on this controller, you must specify one of the following: " +
             "--standalone, --initial-controllers, or --no-initial-controllers.");
-        }
-      }
     }
     Option(namespace.getList("add_scram")).
       foreach(scramArgs => formatter.setScramArguments(scramArgs.asInstanceOf[util.List[String]]))

core/src/test/java/kafka/server/ReconfigurableQuorumIntegrationTest.java

@@ -84,9 +84,8 @@ public void testCreateAndDestroyReconfigurableCluster() throws Exception {
             new TestKitNodes.Builder().
                 setNumBrokerNodes(1).
                 setNumControllerNodes(1).
-                setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
                 build()
-        ).build()) {
+        ).setStandalone(true).build()) {
             cluster.format();
             cluster.startup();
             try (Admin admin = Admin.create(cluster.clientProperties())) {
@@ -108,13 +107,23 @@ static Map<Integer, Uuid> findVoterDirs(Admin admin) throws Exception {
 
     @Test
     public void testRemoveController() throws Exception {
-        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
-            new TestKitNodes.Builder().
-                setNumBrokerNodes(1).
-                setNumControllerNodes(3).
-                setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
-                build()
-        ).build()) {
+        final var nodes = new TestKitNodes.Builder().
+            setNumBrokerNodes(1).
+            setNumControllerNodes(3).
+            build();
+
+        final Map<Integer, Uuid> initialVoters = new HashMap<>();
+        for (final var controllerNode : nodes.controllerNodes().values()) {
+            initialVoters.put(
+                controllerNode.id(),
+                controllerNode.metadataDirectoryId()
+            );
+        }
+
+        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(nodes).
+            setInitialVoterSet(initialVoters).
+            build()
+        ) {
             cluster.format();
             cluster.startup();
             try (Admin admin = Admin.create(cluster.clientProperties())) {
@@ -133,12 +142,22 @@ public void testRemoveController() throws Exception {
 
     @Test
     public void testRemoveAndAddSameController() throws Exception {
-        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
-            new TestKitNodes.Builder().
-                setNumBrokerNodes(1).
-                setNumControllerNodes(4).
-                setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
-                build()).build()
+        final var nodes = new TestKitNodes.Builder().
+            setNumBrokerNodes(1).
+            setNumControllerNodes(4).
+            build();
+
+        final Map<Integer, Uuid> initialVoters = new HashMap<>();
+        for (final var controllerNode : nodes.controllerNodes().values()) {
+            initialVoters.put(
+                controllerNode.id(),
+                controllerNode.metadataDirectoryId()
+            );
+        }
+
+        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(nodes).
+            setInitialVoterSet(initialVoters).
+            build()
         ) {
             cluster.format();
             cluster.startup();
@@ -173,7 +192,6 @@ public void testControllersAutoJoinStandaloneVoter() throws Exception {
         final var nodes = new TestKitNodes.Builder().
             setNumBrokerNodes(1).
             setNumControllerNodes(3).
-            setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
             build();
         try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(nodes).
             setConfigProp(QuorumConfig.QUORUM_AUTO_JOIN_ENABLE_CONFIG, true).
@@ -199,7 +217,6 @@ public void testNewVoterAutoRemovesAndAdds() throws Exception {
         final var nodes = new TestKitNodes.Builder().
             setNumBrokerNodes(1).
             setNumControllerNodes(3).
-            setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
             build();
 
         // Configure the initial voters with one voter having a different directory ID.

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

@@ -605,19 +605,14 @@ Found problem:
               Seq("--release-version", "3.9-IV0"))).getMessage)
   }
 
-  @ParameterizedTest
-  @ValueSource(booleans = Array(false, true))
-  def testFormatWithNoInitialControllersSucceedsOnController(setKraftVersionFeature: Boolean): Unit = {
+  @Test
+  def testFormatWithNoInitialControllersSucceedsOnController(): Unit = {
     val availableDirs = Seq(TestUtils.tempDir())
     val properties = new Properties()
     properties.putAll(defaultDynamicQuorumProperties)
     properties.setProperty("log.dirs", availableDirs.mkString(","))
     val stream = new ByteArrayOutputStream()
     val arguments = ListBuffer[String]("--release-version", "3.9-IV0", "--no-initial-controllers")
-    if (setKraftVersionFeature) {
-      arguments += "--feature"
-      arguments += "kraft.version=1"
-    }
     assertEquals(0, runFormatCommand(stream, properties, arguments.toSeq))
     assertTrue(stream.toString().
       contains("Formatting metadata directory %s".format(availableDirs.head)),

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

@@ -131,7 +131,6 @@ public class Formatter {
      * The initial KIP-853 voters.
      */
     private Optional<DynamicVoters> initialControllers = Optional.empty();
-    private boolean noInitialControllersFlag = false;
 
     public Formatter setPrintStream(PrintStream printStream) {
         this.printStream = printStream;
@@ -217,17 +216,12 @@ public Formatter setInitialControllers(DynamicVoters initialControllers) {
         return this;
     }
 
-    public Formatter setNoInitialControllersFlag(boolean noInitialControllersFlag) {
-        this.noInitialControllersFlag = noInitialControllersFlag;
-        return this;
-    }
-
     public Optional<DynamicVoters> initialVoters() {
         return initialControllers;
     }
 
     boolean hasDynamicQuorum() {
-        return initialControllers.isPresent() || noInitialControllersFlag;
+        return initialControllers.isPresent();
     }
 
     public BootstrapMetadata bootstrapMetadata() {
@@ -337,7 +331,7 @@ Map<String, Short> calculateEffectiveFeatureLevels() {
     /**
      * Calculate the effective feature level for kraft.version. In order to keep existing
      * command-line invocations of StorageTool working, we default this to 0 if no dynamic
-     * voter quorum arguments were provided. As a convenience, if dynamic voter quorum arguments
+     * voter quorum arguments were provided. As a convenience, if --standalone or --initial-voters
      * were passed, we set the latest kraft.version. (Currently there is only 1 non-zero version).
      *
      * @param configuredKRaftVersionLevel   The configured level for kraft.version
@@ -348,20 +342,17 @@ private short effectiveKRaftFeatureLevel(Optional<Short> configuredKRaftVersionL
             if (configuredKRaftVersionLevel.get() == 0) {
                 if (hasDynamicQuorum()) {
                     throw new FormatterException(
-                        "Cannot set kraft.version to " +
-                        configuredKRaftVersionLevel.get() +
-                        " if one of the flags --standalone, --initial-controllers, or --no-initial-controllers is used. " +
+                        "Cannot set kraft.version to " + configuredKRaftVersionLevel.get() +
+                        " if one of the flags --standalone or --initial-controllers is used. " +
                         "For dynamic controllers support, try removing the --feature flag for kraft.version."
                     );
                 }
             } else {
                 if (!hasDynamicQuorum()) {
                     throw new FormatterException(
-                        "Cannot set kraft.version to " +
-                        configuredKRaftVersionLevel.get() +
-                        " 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."
+                        "Cannot set kraft.version to " + configuredKRaftVersionLevel.get() +
+                        " unless one of the flags --standalone or --initial-controllers is used. " +
+                        "For dynamic controllers support, try using one of --standalone or --initial-controllers."
                     );
                 }
             }

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

@@ -462,9 +462,8 @@ public void testFormatWithInitialVotersFailsWithOlderKraftVersion() throws Excep
                     parse("1@localhost:8020:4znU-ou9Taa06bmEJxsjnw"));
             assertTrue(formatter1.formatter.hasDynamicQuorum());
             assertEquals(
-                "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.",
+                "Cannot set kraft.version to 0 if one of the flags --standalone or --initial-controllers " +
+                "is used. For dynamic controllers support, try removing the --feature flag for kraft.version.",
                 assertThrows(FormatterException.class, formatter1.formatter::run).getMessage()
             );
         }
@@ -478,9 +477,8 @@ public void testFormatWithoutInitialVotersFailsWithNewerKraftVersion() throws Ex
             formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
             assertFalse(formatter1.formatter.hasDynamicQuorum());
             assertEquals(
-                "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.",
+                "Cannot set kraft.version to 1 unless one of the flags --standalone or --initial-controllers " +
+                "is used. For dynamic controllers support, try using one of --standalone or --initial-controllers.",
                 assertThrows(FormatterException.class, formatter1.formatter::run).getMessage()
             );
         }
@@ -502,6 +500,17 @@ public void testFormatWithInitialVotersFailsWithOlderMetadataVersion() throws Ex
         }
     }
 
+    @Test
+    public void testFormatWithNoInitialVotersFailsWithOlderMetadataVersion() throws Exception {
+        try (TestEnv testEnv = new TestEnv(2)) {
+            FormatterContext formatter1 = testEnv.newFormatter();
+            formatter1.formatter.setReleaseVersion(MetadataVersion.IBP_3_8_IV0);
+            // This MV does not support kraft.version = 1
+            formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
+            formatter1.formatter.run();
+        }
+    }
+
     private static Stream<Arguments> elrTestMetadataVersions() {
         return Stream.of(
             MetadataVersion.IBP_3_9_IV0,
@@ -530,20 +539,14 @@ public void testFormatElrEnabledWithMetadataVersions(MetadataVersion metadataVer
         }
     }
 
-    @ParameterizedTest
-    @ValueSource(booleans = {false, true})
-    public void testFormatWithNoInitialControllers(boolean specifyKRaftVersion) throws Exception {
+    @Test
+    public void testFormatWithNoInitialControllers() throws Exception {
         try (TestEnv testEnv = new TestEnv(2)) {
             FormatterContext formatter1 = testEnv.newFormatter();
-            if (specifyKRaftVersion) {
-                formatter1.formatter.setFeatureLevel("kraft.version", (short) 1);
-            }
             formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
-            formatter1.formatter.setNoInitialControllersFlag(true);
-            assertTrue(formatter1.formatter.hasDynamicQuorum());
-
+            assertFalse(formatter1.formatter.hasDynamicQuorum());
             formatter1.formatter.run();
-            assertEquals((short) 1, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 0));
+            assertEquals((short) 0, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 1));
             assertEquals(List.of(
                     "Bootstrap metadata: " + formatter1.formatter.bootstrapMetadata(),
                     String.format("Formatting data directory %s with %s %s.",
@@ -571,31 +574,23 @@ public void testFormatWithoutNoInitialControllersFailsWithNewerKraftVersion() th
             FormatterContext formatter1 = testEnv.newFormatter();
             formatter1.formatter.setFeatureLevel("kraft.version", (short) 1);
             formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
-            formatter1.formatter.setNoInitialControllersFlag(false);
             assertFalse(formatter1.formatter.hasDynamicQuorum());
             assertEquals(
-                "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.",
+                "Cannot set kraft.version to 1 unless one of the flags --standalone or --initial-controllers " +
+                "is used. For dynamic controllers support, try using one of --standalone or --initial-controllers.",
                 assertThrows(FormatterException.class, formatter1.formatter::run).getMessage()
             );
         }
     }
 
     @Test
-    public void testFormatWithNoInitialControllersFailsWithOlderKraftVersion() throws Exception {
+    public void testFormatWithNoInitialControllersPassesWithOlderKraftVersion() throws Exception {
         try (TestEnv testEnv = new TestEnv(2)) {
             FormatterContext formatter1 = testEnv.newFormatter();
             formatter1.formatter.setFeatureLevel("kraft.version", (short) 0);
             formatter1.formatter.setUnstableFeatureVersionsEnabled(true);
-            formatter1.formatter.setNoInitialControllersFlag(true);
-            assertTrue(formatter1.formatter.hasDynamicQuorum());
-            assertEquals(
-                "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()
-            );
+            assertFalse(formatter1.formatter.hasDynamicQuorum());
+            formatter1.formatter.run();
         }
     }
 }