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

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

@@ -149,8 +149,9 @@ object StorageTool extends Logging {
       })
     val initialControllers = namespace.getString("initial_controllers")
     val isStandalone = namespace.getBoolean("standalone")
-    if (!config.quorumConfig.voters().isEmpty &&
-      (Option(initialControllers).isDefined || isStandalone)) {
+    val staticVotersEmpty = config.quorumConfig.voters().isEmpty
+    formatter.setHasDynamicQuorum(staticVotersEmpty)
+    if (!staticVotersEmpty && (Option(initialControllers).isDefined || isStandalone)) {
       throw new TerseFailure("You cannot specify " +
         QuorumConfig.QUORUM_VOTERS_CONFIG + " and format the node " +
         "with --initial-controllers or --standalone. " +
@@ -163,16 +164,12 @@ 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) &&
+      staticVotersEmpty && 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]]))
@@ -342,18 +339,21 @@ object StorageTool extends Logging {
 
     val reconfigurableQuorumOptions = formatParser.addMutuallyExclusiveGroup()
     reconfigurableQuorumOptions.addArgument("--standalone", "-s")
-      .help("Used to initialize a controller as a single-node dynamic quorum.")
+      .help("Used to initialize a controller as a single-node dynamic quorum. When setting this flag, " +
+        "the controller.quorum.voters config must not be set, and controller.quorum.bootstrap.servers is set instead.")
       .action(storeTrue())
 
     reconfigurableQuorumOptions.addArgument("--no-initial-controllers", "-N")
-      .help("Used to initialize a server without a dynamic quorum topology.")
+      .help("Used to initialize a server without a dynamic quorum topology. When setting this flag, " +
+        "the controller.quorum.voters config should not be set, and controller.quorum.bootstrap.servers is set instead.")
       .action(storeTrue())
 
     reconfigurableQuorumOptions.addArgument("--initial-controllers", "-I")
       .help("Used to initialize a server with a specific dynamic quorum topology. The argument " +
         "is a comma-separated list of id@hostname:port:directory. The same values must be used to " +
         "format all nodes. For example:\n0@example.com:8082:JEXY6aqzQY-32P5TStzaFg,1@example.com:8083:" +
-        "MvDxzVmcRsaTz33bUuRU6A,2@example.com:8084:07R5amHmR32VDA6jHkGbTA\n")
+        "MvDxzVmcRsaTz33bUuRU6A,2@example.com:8084:07R5amHmR32VDA6jHkGbTA\n. When setting this flag, " +
+        "the controller.quorum.voters config must not be set, and controller.quorum.bootstrap.servers is set instead.")
       .action(store())
   }
 

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/integration/kafka/server/KRaftClusterTest.scala

@@ -1035,8 +1035,7 @@ class KRaftClusterTest {
     val cluster = new KafkaClusterTestKit.Builder(
       new TestKitNodes.Builder().
         setNumBrokerNodes(1).
-        setNumControllerNodes(1).
-        setFeature(KRaftVersion.FEATURE_NAME, 1.toShort).build()).build()
+        setNumControllerNodes(1).build()).setStandalone(true).build()
     try {
       cluster.format()
       cluster.startup()

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

@@ -351,7 +351,7 @@ Found problem:
     val properties = new Properties()
     properties.putAll(defaultDynamicQuorumProperties)
     properties.setProperty("log.dirs", availableDirs.mkString(","))
-    assertEquals("No feature:kraft.version with feature level 999",
+    assertEquals("Unknown KRaft feature level: 999",
       assertThrows(classOf[IllegalArgumentException], () =>
         runFormatCommand(new ByteArrayOutputStream(), properties,
           Seq("--feature", "kraft.version=999", "--standalone"))).getMessage)
@@ -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,7 @@ public class Formatter {
      * The initial KIP-853 voters.
      */
     private Optional<DynamicVoters> initialControllers = Optional.empty();
-    private boolean noInitialControllersFlag = false;
+    private boolean hasDynamicQuorum = false;
 
     public Formatter setPrintStream(PrintStream printStream) {
         this.printStream = printStream;
@@ -217,8 +217,8 @@ public Formatter setInitialControllers(DynamicVoters initialControllers) {
         return this;
     }
 
-    public Formatter setNoInitialControllersFlag(boolean noInitialControllersFlag) {
-        this.noInitialControllersFlag = noInitialControllersFlag;
+    public Formatter setHasDynamicQuorum(boolean staticVotersEmpty) {
+        this.hasDynamicQuorum = staticVotersEmpty;
         return this;
     }
 
@@ -227,7 +227,7 @@ public Optional<DynamicVoters> initialVoters() {
     }
 
     boolean hasDynamicQuorum() {
-        return initialControllers.isPresent() || noInitialControllersFlag;
+        return hasDynamicQuorum;
     }
 
     public BootstrapMetadata bootstrapMetadata() {
@@ -306,15 +306,15 @@ Map<String, Short> calculateEffectiveFeatureLevels() {
                             ". Supported features are: " + String.join(", ", nameToSupportedFeature.keySet()));
                 }
             }
-            newFeatureLevels.put(featureName, level);
+            if (!featureName.equals(KRaftVersion.FEATURE_NAME)) {
+                newFeatureLevels.put(featureName, level);
+            }
         }
         newFeatureLevels.put(MetadataVersion.FEATURE_NAME, releaseVersion.featureLevel());
         // Add default values for features that were not specified.
         supportedFeatures.forEach(supportedFeature -> {
-            if (supportedFeature.featureName().equals(KRaftVersion.FEATURE_NAME)) {
-                newFeatureLevels.put(KRaftVersion.FEATURE_NAME, effectiveKRaftFeatureLevel(
-                    Optional.ofNullable(newFeatureLevels.get(KRaftVersion.FEATURE_NAME))));
-            } else if (!newFeatureLevels.containsKey(supportedFeature.featureName())) {
+            if (!newFeatureLevels.containsKey(supportedFeature.featureName()) &&
+                !supportedFeature.featureName().equals(KRaftVersion.FEATURE_NAME)) {
                 newFeatureLevels.put(supportedFeature.featureName(),
                     supportedFeature.defaultLevel(releaseVersion));
             }
@@ -331,14 +331,18 @@ Map<String, Short> calculateEffectiveFeatureLevels() {
                 Feature.validateVersion(featureVersion, newFeatureLevels);
             }
         }
+        // Only add kraft.version after validation, because it only depends
+        // on format arguments, not the release version.
+        newFeatureLevels.put(KRaftVersion.FEATURE_NAME, effectiveKRaftFeatureLevel(
+            Optional.ofNullable(featureLevels.get(KRaftVersion.FEATURE_NAME))));
         return newFeatureLevels;
     }
 
     /**
      * 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
-     * were passed, we set the latest kraft.version. (Currently there is only 1 non-zero version).
+     * voter quorum arguments were provided. As a convenience, if the static voters config is
+     * empty, we set the latest kraft.version. (Currently there is only 1 non-zero version).
      *
      * @param configuredKRaftVersionLevel   The configured level for kraft.version
      * @return                              The effective feature level.
@@ -348,20 +352,20 @@ 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 controller.quorum.voters is empty and 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."
                     );
                 }
             } 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 controller.quorum.voters is empty and 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 and removing controller.quorum.voters."
                     );
                 }
             }