Skip to content

Commit f5473df

Browse files
kevin-wu24kevin-wu24
committed
KAFKA-19719 --no-initial-controllers should not assume kraft.version=1 (apache#20604)
``` commit ec37eb5 (HEAD -> KAFKA-19719-cherry-pick-41, origin/KAFKA-19719-cherry-pick-41) Author: Kevin Wu <[email protected]> Date: Thu Sep 25 11:56:16 2025 -0500 KAFKA-19719: --no-initial-controllers should not assume kraft.version=1 (apache#20551) Just because a controller node sets --no-initial-controllers flag does not mean it is necessarily running kraft.version=1. The more precise meaning is that the controller node being formatted does not know what kraft version the cluster should be in, and therefore it is only safe to assume kraft.version=0. Only by setting --standalone,--initial-controllers, or --no-initial-controllers AND not specifying the controller.quorum.voters static config, is it known kraft.version > 0. For example, it is a valid configuration (although confusing) to run a static quorum defined by controller.quorum.voters but have all the controllers format with --no-initial-controllers. In this case, specifying --no-initial-controllers alongside a metadata version that does not support kraft.version=1 causes formatting to fail, which is does not support kraft.version=1 causes formatting to fail, which is a regression. Additionally, the formatter should not check the kraft.version against the release version, since kraft.version does not actually depend on any release version. It should only check the kraft.version against the static voters config/format arguments. This PR also cleans up the integration test framework to match the semantics of formatting an actual cluster. Reviewers: TengYao Chi <[email protected]>, Kuan-Po Tseng <[email protected]>, Chia-Ping Tsai <[email protected]>, José Armando García Sancio <[email protected]> Conflicts: core/src/main/scala/kafka/tools/StorageTool.scala Minor conflicts. Keep changes from cherry-pick. core/src/test/java/kafka/server/ReconfigurableQuorumIntegrationTest.java Remove auto-join tests, since 4.1 does not support it. docs/ops.html Keep docs section from cherry-pick. metadata/src/test/java/org/apache/kafka/metadata/storage/FormatterTest.java Minor conflicts. Keep cherry-picked changes. test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java Conflicts due to integration test framework changes. Keep new changes. commit 02d58b1 (upstream/4.1) ``` Reviewers: Chia-Ping Tsai <[email protected]> Conflicts: core/src/test/java/kafka/server/ReconfigurableQuorumIntegrationTest.java docs/ops.html metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java metadata/src/test/java/org/apache/kafka/metadata/storage/FormatterTest.java server-common/src/main/java/org/apache/kafka/server/common/KRaftVersion.java test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java All minor conflicts. Keep cherry-picked changes.
1 parent 8248d1d commit f5473df

File tree

10 files changed

+279
-185
lines changed

10 files changed

+279
-185
lines changed

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

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -135,21 +135,30 @@ object StorageTool extends Logging {
135135
featureNamesAndLevels(_).foreachEntry {
136136
(k, v) => formatter.setFeatureLevel(k, v)
137137
})
138-
Option(namespace.getString("initial_controllers")).
138+
val initialControllers = namespace.getString("initial_controllers")
139+
val isStandalone = namespace.getBoolean("standalone")
140+
val staticVotersEmpty = config.quorumConfig.voters().isEmpty
141+
formatter.setHasDynamicQuorum(staticVotersEmpty)
142+
if (!staticVotersEmpty && (Option(initialControllers).isDefined || isStandalone)) {
143+
throw new TerseFailure("You cannot specify " +
144+
QuorumConfig.QUORUM_VOTERS_CONFIG + " and format the node " +
145+
"with --initial-controllers or --standalone. " +
146+
"If you want to use dynamic quorum, please remove " +
147+
QuorumConfig.QUORUM_VOTERS_CONFIG + " and specify " +
148+
QuorumConfig.QUORUM_BOOTSTRAP_SERVERS_CONFIG + " instead.")
149+
}
150+
Option(initialControllers).
139151
foreach(v => formatter.setInitialControllers(DynamicVoters.parse(v)))
140-
if (namespace.getBoolean("standalone")) {
152+
if (isStandalone) {
141153
formatter.setInitialControllers(createStandaloneDynamicVoters(config))
142154
}
143-
if (namespace.getBoolean("no_initial_controllers")) {
144-
formatter.setNoInitialControllersFlag(true)
145-
} else {
146-
if (config.processRoles.contains(ProcessRole.ControllerRole)) {
147-
if (config.quorumConfig.voters().isEmpty && formatter.initialVoters().isEmpty) {
155+
if (!namespace.getBoolean("no_initial_controllers") &&
156+
config.processRoles.contains(ProcessRole.ControllerRole) &&
157+
staticVotersEmpty &&
158+
formatter.initialVoters().isEmpty) {
148159
throw new TerseFailure("Because " + QuorumConfig.QUORUM_VOTERS_CONFIG +
149160
" is not set on this controller, you must specify one of the following: " +
150161
"--standalone, --initial-controllers, or --no-initial-controllers.");
151-
}
152-
}
153162
}
154163
Option(namespace.getList("add_scram")).
155164
foreach(scramArgs => formatter.setScramArguments(scramArgs.asInstanceOf[util.List[String]]))
@@ -319,18 +328,21 @@ object StorageTool extends Logging {
319328

320329
val reconfigurableQuorumOptions = formatParser.addMutuallyExclusiveGroup()
321330
reconfigurableQuorumOptions.addArgument("--standalone", "-s")
322-
.help("Used to initialize a controller as a single-node dynamic quorum.")
331+
.help("Used to initialize a controller as a single-node dynamic quorum. When setting this flag, " +
332+
"the controller.quorum.voters config must not be set, and controller.quorum.bootstrap.servers is set instead.")
323333
.action(storeTrue())
324334

325335
reconfigurableQuorumOptions.addArgument("--no-initial-controllers", "-N")
326-
.help("Used to initialize a server without a dynamic quorum topology.")
336+
.help("Used to initialize a server without specifying a dynamic quorum. When setting this flag, " +
337+
"the controller.quorum.voters config should not be set, and controller.quorum.bootstrap.servers is set instead.")
327338
.action(storeTrue())
328339

329340
reconfigurableQuorumOptions.addArgument("--initial-controllers", "-I")
330-
.help("Used to initialize a server with a specific dynamic quorum topology. The argument " +
341+
.help("Used to initialize a server with the specified dynamic quorum. The argument " +
331342
"is a comma-separated list of id@hostname:port:directory. The same values must be used to " +
332343
"format all nodes. For example:\n[email protected]:8082:JEXY6aqzQY-32P5TStzaFg,[email protected]:8083:" +
333-
"MvDxzVmcRsaTz33bUuRU6A,[email protected]:8084:07R5amHmR32VDA6jHkGbTA\n")
344+
"MvDxzVmcRsaTz33bUuRU6A,[email protected]:8084:07R5amHmR32VDA6jHkGbTA\n. When setting this flag, " +
345+
"the controller.quorum.voters config must not be set, and controller.quorum.bootstrap.servers is set instead.")
334346
.action(store())
335347
}
336348

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

Lines changed: 35 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131

3232
import java.util.Arrays;
3333
import java.util.Collections;
34+
import java.util.HashMap;
3435
import java.util.HashSet;
3536
import java.util.Map;
3637
import java.util.TreeMap;
@@ -83,9 +84,8 @@ public void testCreateAndDestroyReconfigurableCluster() throws Exception {
8384
new TestKitNodes.Builder().
8485
setNumBrokerNodes(1).
8586
setNumControllerNodes(1).
86-
setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
8787
build()
88-
).build()) {
88+
).setStandalone(true).build()) {
8989
cluster.format();
9090
cluster.startup();
9191
try (Admin admin = Admin.create(cluster.clientProperties())) {
@@ -107,13 +107,23 @@ static Map<Integer, Uuid> findVoterDirs(Admin admin) throws Exception {
107107

108108
@Test
109109
public void testRemoveController() throws Exception {
110-
try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
111-
new TestKitNodes.Builder().
112-
setNumBrokerNodes(1).
113-
setNumControllerNodes(3).
114-
setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
115-
build()
116-
).build()) {
110+
final var nodes = new TestKitNodes.Builder().
111+
setNumBrokerNodes(1).
112+
setNumControllerNodes(3).
113+
build();
114+
115+
final Map<Integer, Uuid> initialVoters = new HashMap<>();
116+
for (final var controllerNode : nodes.controllerNodes().values()) {
117+
initialVoters.put(
118+
controllerNode.id(),
119+
controllerNode.metadataDirectoryId()
120+
);
121+
}
122+
123+
try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(nodes).
124+
setInitialVoterSet(initialVoters).
125+
build()
126+
) {
117127
cluster.format();
118128
cluster.startup();
119129
try (Admin admin = Admin.create(cluster.clientProperties())) {
@@ -132,12 +142,22 @@ public void testRemoveController() throws Exception {
132142

133143
@Test
134144
public void testRemoveAndAddSameController() throws Exception {
135-
try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
136-
new TestKitNodes.Builder().
137-
setNumBrokerNodes(1).
138-
setNumControllerNodes(4).
139-
setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel()).
140-
build()).build()
145+
final var nodes = new TestKitNodes.Builder().
146+
setNumBrokerNodes(1).
147+
setNumControllerNodes(4).
148+
build();
149+
150+
final Map<Integer, Uuid> initialVoters = new HashMap<>();
151+
for (final var controllerNode : nodes.controllerNodes().values()) {
152+
initialVoters.put(
153+
controllerNode.id(),
154+
controllerNode.metadataDirectoryId()
155+
);
156+
}
157+
158+
try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(nodes).
159+
setInitialVoterSet(initialVoters).
160+
build()
141161
) {
142162
cluster.format();
143163
cluster.startup();

‎core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1013,8 +1013,7 @@ class KRaftClusterTest {
10131013
val cluster = new KafkaClusterTestKit.Builder(
10141014
new TestKitNodes.Builder().
10151015
setNumBrokerNodes(1).
1016-
setNumControllerNodes(1).
1017-
setFeature(KRaftVersion.FEATURE_NAME, 1.toShort).build()).build()
1016+
setNumControllerNodes(1).build()).setStandalone(true).build()
10181017
try {
10191018
cluster.format()
10201019
cluster.startup()

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -375,7 +375,10 @@ Found problem:
375375
def testFormatWithStandaloneFlagOnBrokerFails(): Unit = {
376376
val availableDirs = Seq(TestUtils.tempDir())
377377
val properties = new Properties()
378-
properties.putAll(defaultStaticQuorumProperties)
378+
properties.setProperty("process.roles", "broker")
379+
properties.setProperty("node.id", "0")
380+
properties.setProperty("controller.listener.names", "CONTROLLER")
381+
properties.setProperty("controller.quorum.bootstrap.servers", "localhost:9093")
379382
properties.setProperty("log.dirs", availableDirs.mkString(","))
380383
val stream = new ByteArrayOutputStream()
381384
val arguments = ListBuffer[String]("--release-version", "3.9-IV0", "--standalone")
@@ -458,19 +461,14 @@ Found problem:
458461
Seq("--release-version", "3.9-IV0"))).getMessage)
459462
}
460463

461-
@ParameterizedTest
462-
@ValueSource(booleans = Array(false, true))
463-
def testFormatWithNoInitialControllersSucceedsOnController(setKraftVersionFeature: Boolean): Unit = {
464+
@Test
465+
def testFormatWithNoInitialControllersSucceedsOnController(): Unit = {
464466
val availableDirs = Seq(TestUtils.tempDir())
465467
val properties = new Properties()
466468
properties.putAll(defaultDynamicQuorumProperties)
467469
properties.setProperty("log.dirs", availableDirs.mkString(","))
468470
val stream = new ByteArrayOutputStream()
469471
val arguments = ListBuffer[String]("--release-version", "3.9-IV0", "--no-initial-controllers")
470-
if (setKraftVersionFeature) {
471-
arguments += "--feature"
472-
arguments += "kraft.version=1"
473-
}
474472
assertEquals(0, runFormatCommand(stream, properties, arguments.toSeq))
475473
assertTrue(stream.toString().
476474
contains("Formatting metadata directory %s".format(availableDirs.head)),

‎docs/ops.html‎

Lines changed: 23 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -3869,45 +3869,29 @@ <h5 class="anchor-heading"><a id="static_versus_dynamic_kraft_quorums" class="an
38693869
If you are not sure whether you are using static or dynamic quorums, you can determine this by
38703870
running something like the following:<p>
38713871

3872-
<pre><code class="language-bash">
3873-
$ bin/kafka-features.sh --bootstrap-controller localhost:9093 describe
3874-
</code></pre><p>
3875-
3876-
If the <code>kraft.version</code> field is level 0 or absent, you are using a static quorum. If
3877-
it is 1 or above, you are using a dynamic quorum. For example, here is an example of a static
3878-
quorum:<p/>
3879-
<pre><code class="language-bash">
3880-
Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 0 Epoch: 5
3881-
Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5
3882-
</code></pre><p/>
3883-
3884-
Here is another example of a static quorum:<p/>
3885-
<pre><code class="language-bash">
3886-
Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.8-IV0 FinalizedVersionLevel: 3.8-IV0 Epoch: 5
3887-
</code></pre><p/>
3888-
3889-
Here is an example of a dynamic quorum:<p/>
3890-
<pre><code class="language-bash">
3891-
Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 1 Epoch: 5
3892-
Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5
3893-
</code></pre><p/>
3894-
3895-
The static versus dynamic nature of the quorum is determined at the time of formatting.
3896-
Specifically, the quorum will be formatted as dynamic if <code>controller.quorum.voters</code> is
3897-
<b>not</b> present, and if the software version is Apache Kafka 3.9 or newer. If you have
3898-
followed the instructions earlier in this document, you will get a dynamic quorum.<p>
3899-
3900-
If you would like the formatting process to fail if a dynamic quorum cannot be achieved, format your
3901-
controllers using the <code>--feature kraft.version=1</code>. (Note that you should not supply
3902-
this flag when formatting brokers -- only when formatting controllers.)<p>
3903-
3904-
<pre><code class="language-bash">
3905-
$ bin/kafka-storage.sh format -t KAFKA_CLUSTER_ID --feature kraft.version=1 -c controller_static.properties
3906-
Cannot set kraft.version to 1 unless KIP-853 configuration is present. Try removing the --feature flag for kraft.version.
3907-
</code></pre><p>
3908-
3909-
Note: Currently it is <b>not</b> possible to convert clusters using a static controller quorum to
3910-
use a dynamic controller quorum. This function will be supported in the future release.
3872+
<pre><code class="language-bash">$ bin/kafka-features.sh --bootstrap-controller localhost:9093 describe</code></pre>
3873+
<p>
3874+
If the <code>kraft.version</code> field is level 0 or absent, you are using a static quorum. If
3875+
it is 1 or above, you are using a dynamic quorum. For example, here is an example of a static
3876+
quorum:<p>
3877+
<pre><code class="language-bash">Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 0 Epoch: 5
3878+
Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5</code></pre>
3879+
<p>
3880+
Here is another example of a static quorum:<p>
3881+
<pre><code class="language-bash">Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.8-IV0 FinalizedVersionLevel: 3.8-IV0 Epoch: 5</code></pre>
3882+
<p>
3883+
Here is an example of a dynamic quorum:<p>
3884+
<pre><code class="language-bash">Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 1 Epoch: 5
3885+
Feature: metadata.version SupportedMinVersion: 3.3-IV3 SupportedMaxVersion: 3.9-IV0 FinalizedVersionLevel: 3.9-IV0 Epoch: 5</code></pre>
3886+
<p>
3887+
The static versus dynamic nature of the quorum is determined at the time of formatting.
3888+
Specifically, the quorum will be formatted as dynamic if <code>controller.quorum.voters</code> is
3889+
<b>not</b> present, and one of --standalone, --initial-controllers, or --no-initial-controllers is set.
3890+
If you have followed the instructions earlier in this document, you will get a dynamic quorum.
3891+
<p>
3892+
3893+
Note: Currently it is <b>not</b> possible to convert clusters using a static controller quorum to
3894+
use a dynamic controller quorum. This function will be supported in the future release.
39113895

39123896
<h5 class="anchor-heading"><a id="kraft_reconfig_add" class="anchor-link"></a><a href="#kraft_reconfig_add">Add New Controller</a></h5>
39133897
If a dynamic controller cluster already exists, it can be expanded by first provisioning a new controller using the <a href="#kraft_storage_observers">kafka-storage.sh tool</a> and starting the controller.

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

Lines changed: 18 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ public class Formatter {
132132
* The initial KIP-853 voters.
133133
*/
134134
private Optional<DynamicVoters> initialControllers = Optional.empty();
135-
private boolean noInitialControllersFlag = false;
135+
private boolean hasDynamicQuorum = false;
136136

137137
public Formatter setPrintStream(PrintStream printStream) {
138138
this.printStream = printStream;
@@ -218,8 +218,8 @@ public Formatter setInitialControllers(DynamicVoters initialControllers) {
218218
return this;
219219
}
220220

221-
public Formatter setNoInitialControllersFlag(boolean noInitialControllersFlag) {
222-
this.noInitialControllersFlag = noInitialControllersFlag;
221+
public Formatter setHasDynamicQuorum(boolean hasDynamicQuorum) {
222+
this.hasDynamicQuorum = hasDynamicQuorum;
223223
return this;
224224
}
225225

@@ -228,7 +228,7 @@ public Optional<DynamicVoters> initialVoters() {
228228
}
229229

230230
boolean hasDynamicQuorum() {
231-
return initialControllers.isPresent() || noInitialControllersFlag;
231+
return hasDynamicQuorum;
232232
}
233233

234234
public BootstrapMetadata bootstrapMetadata() {
@@ -338,8 +338,8 @@ Map<String, Short> calculateEffectiveFeatureLevels() {
338338
/**
339339
* Calculate the effective feature level for kraft.version. In order to keep existing
340340
* command-line invocations of StorageTool working, we default this to 0 if no dynamic
341-
* voter quorum arguments were provided. As a convenience, if dynamic voter quorum arguments
342-
* were passed, we set the latest kraft.version. (Currently there is only 1 non-zero version).
341+
* voter quorum arguments were provided. As a convenience, if the static voters config is
342+
* empty, we set the latest kraft.version. (Currently there is only 1 non-zero version).
343343
*
344344
* @param configuredKRaftVersionLevel The configured level for kraft.version
345345
* @return The effective feature level.
@@ -348,15 +348,21 @@ private short effectiveKRaftFeatureLevel(Optional<Short> configuredKRaftVersionL
348348
if (configuredKRaftVersionLevel.isPresent()) {
349349
if (configuredKRaftVersionLevel.get() == 0) {
350350
if (hasDynamicQuorum()) {
351-
throw new FormatterException("Cannot set kraft.version to " +
352-
configuredKRaftVersionLevel.get() + " if KIP-853 configuration is present. " +
353-
"Try removing the --feature flag for kraft.version.");
351+
throw new FormatterException(
352+
"Cannot set kraft.version to 0 if controller.quorum.voters is empty and one of the flags " +
353+
"--standalone, --initial-controllers, or --no-initial-controllers is used. For dynamic " +
354+
"controllers support, try removing the --feature flag for kraft.version."
355+
);
354356
}
355357
} else {
356358
if (!hasDynamicQuorum()) {
357-
throw new FormatterException("Cannot set kraft.version to " +
358-
configuredKRaftVersionLevel.get() + " unless KIP-853 configuration is present. " +
359-
"Try removing the --feature flag for kraft.version.");
359+
throw new FormatterException(
360+
"Cannot set kraft.version to " + configuredKRaftVersionLevel.get() +
361+
" unless controller.quorum.voters is empty and one of the flags --standalone, " +
362+
"--initial-controllers, or --no-initial-controllers is used. " +
363+
"For dynamic controllers support, try using one of --standalone, --initial-controllers, " +
364+
"or --no-initial-controllers and removing controller.quorum.voters."
365+
);
360366
}
361367
}
362368
return configuredKRaftVersionLevel.get();

0 commit comments

Comments
 (0)