root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-18028 the effective kraft version of --no-initial-controllers should be 1 rather than 0 (#17836) 

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
PoAn Yang 2024-11-27 01:45:11 +08:00 committed by GitHub
parent b42efc7dc2
commit 98d47f47ef
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 114 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
core/src/main/scala/kafka/tools/StorageTool.scala
View File

@ -143,17 +143,17 @@ object StorageTool extends Logging {
if (namespace.getBoolean("standalone")) {
formatter.setInitialControllers(createStandaloneDynamicVoters(config))
}
if (!namespace.getBoolean("no_initial_controllers")) {
if (namespace.getBoolean("no_initial_controllers")) {
formatter.setNoInitialControllersFlag(true)
} else {
if (config.processRoles.contains(ProcessRole.ControllerRole)) {
if (config.quorumConfig.voters().isEmpty) {
if (formatter.initialVoters().isEmpty()) {
if (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]]))
configToLogDirectories(config).foreach(formatter.addDirectory(_))

36
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala
View File

@ -483,20 +483,48 @@ Found problem:
Seq("--release-version", "3.9-IV0"))).getMessage)
}
@Test
def testFormatWithNoInitialControllersSucceedsOnController(): Unit = {
@ParameterizedTest
@ValueSource(booleans = Array(false, true))
def testFormatWithNoInitialControllersSucceedsOnController(setKraftVersionFeature: Boolean): Unit = {
val availableDirs = Seq(TestUtils.tempDir())
val properties = new Properties()
properties.putAll(defaultDynamicQuorumProperties)
properties.setProperty("log.dirs", availableDirs.mkString(","))
val stream = new ByteArrayOutputStream()
assertEquals(0, runFormatCommand(stream, properties,
Seq("--no-initial-controllers", "--release-version", "3.9-IV0")))
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)),
"Failed to find content in output: " + stream.toString())
}
@Test
def testFormatWithNoInitialControllersFlagAndStandaloneFlagFails(): Unit = {
val arguments = ListBuffer[String](
"format", "--cluster-id", "XcZZOzUqS4yHOjhMQB6JLQ",
"--release-version", "3.9-IV0",
"--no-initial-controllers", "--standalone")
val exception = assertThrows(classOf[ArgumentParserException], () => StorageTool.parseArguments(arguments.toArray))
assertEquals("argument --standalone/-s: not allowed with argument --no-initial-controllers/-N", exception.getMessage)
}
@Test
def testFormatWithNoInitialControllersFlagAndInitialControllersFlagFails(): Unit = {
val arguments = ListBuffer[String](
"format", "--cluster-id", "XcZZOzUqS4yHOjhMQB6JLQ",
"--release-version", "3.9-IV0",
"--no-initial-controllers", "--initial-controllers",
"0@localhost:8020:K90IZ-0DRNazJ49kCZ1EMQ," +
"1@localhost:8030:aUARLskQTCW4qCZDtS_cwA," +
"2@localhost:8040:2ggvsS4kQb-fSJ_-zC_Ang")
val exception = assertThrows(classOf[ArgumentParserException], () => StorageTool.parseArguments(arguments.toArray))
assertEquals("argument --initial-controllers/-I: not allowed with argument --no-initial-controllers/-N", exception.getMessage)
}
@Test
def testFormatWithoutStaticQuorumSucceedsWithoutInitialControllersOnBroker(): Unit = {
val availableDirs = Seq(TestUtils.tempDir())

12
metadata/src/main/java/org/apache/kafka/metadata/storage/Formatter.java
View File

@ -93,8 +93,10 @@ public class Formatter {
/**
* Maps feature names to the level they will start off with.
*
* Visible for testing.
*/
private Map<String, Short> featureLevels = new TreeMap<>();
protected Map<String, Short> featureLevels = new TreeMap<>();
/**
* The bootstrap metadata used to format the cluster.
@ -130,6 +132,7 @@ 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;
@ -215,12 +218,17 @@ public class Formatter {
return this;
}
public Formatter setNoInitialControllersFlag(boolean noInitialControllersFlag) {
this.noInitialControllersFlag = noInitialControllersFlag;
return this;
}
public Optional<DynamicVoters> initialVoters() {
return initialControllers;
}
boolean hasDynamicQuorum() {
return initialControllers.isPresent();
return initialControllers.isPresent() || noInitialControllersFlag;
}
public BootstrapMetadata bootstrapMetadata() {

65
metadata/src/test/java/org/apache/kafka/metadata/storage/FormatterTest.java
View File

@ -379,6 +379,7 @@ public class FormatterTest {
formatter1.formatter.setInitialControllers(DynamicVoters.
parse("1@localhost:8020:4znU-ou9Taa06bmEJxsjnw"));
formatter1.formatter.run();
assertEquals((short) 1, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 0));
assertEquals(Arrays.asList(
String.format("Formatting data directory %s with %s %s.",
testEnv.directory(1),
@ -446,4 +447,68 @@ public class FormatterTest {
() -> formatter1.formatter.run()).getMessage());
}
}
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testFormatWithNoInitialControllers(boolean specifyKRaftVersion) 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());
formatter1.formatter.run();
assertEquals((short) 1, formatter1.formatter.featureLevels.getOrDefault("kraft.version", (short) 0));
assertEquals(Arrays.asList(
String.format("Formatting data directory %s with %s %s.",
testEnv.directory(1),
MetadataVersion.FEATURE_NAME,
MetadataVersion.latestTesting()),
String.format("Formatting metadata directory %s with %s %s.",
testEnv.directory(0),
MetadataVersion.FEATURE_NAME,
MetadataVersion.latestTesting())),
formatter1.outputLines().stream().sorted().collect(Collectors.toList()));
MetaPropertiesEnsemble ensemble = new MetaPropertiesEnsemble.Loader().
addLogDirs(testEnv.directories).
load();
MetaProperties logDirProps0 = ensemble.logDirProps().get(testEnv.directory(0));
assertNotNull(logDirProps0);
MetaProperties logDirProps1 = ensemble.logDirProps().get(testEnv.directory(1));
assertNotNull(logDirProps1);
}
}
@Test
public void testFormatWithoutNoInitialControllersFailsWithNewerKraftVersion() throws Exception {
try (TestEnv testEnv = new TestEnv(2)) {
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 KIP-853 configuration is present. " +
"Try removing the --feature flag for kraft.version.",
assertThrows(FormatterException.class,
formatter1.formatter::run).getMessage());
}
}
@Test
public void testFormatWithNoInitialControllersFailsWithOlderKraftVersion() 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 KIP-853 configuration is present. " +
"Try removing the --feature flag for kraft.version.",
assertThrows(FormatterException.class,
formatter1.formatter::run).getMessage());
}
}
}