(kafka) branch trunk updated: KAFKA-17543: Enforce that broker.id.generation.enable is not used when migrating to KRaft (#17192)

[cm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new d7a456ed5d5 KAFKA-17543: Enforce that broker.id.generation.enable is not used when migrating to KRaft (#17192)
d7a456ed5d5 is described below

commit d7a456ed5d531465d33cb83f4fafe7f79327d6b8
Author: Colin Patrick McCabe <cm...@apache.org>
AuthorDate: Fri Sep 13 17:24:48 2024 -0700

    KAFKA-17543: Enforce that broker.id.generation.enable is not used when migrating to KRaft (#17192)
    
    Reviewers: Chia-Ping Tsai <ch...@gmail.com>, David Arthur <mu...@gmail.com>
---
 core/src/main/scala/kafka/server/KafkaConfig.scala          |  5 +++++
 core/src/main/scala/kafka/server/KafkaServer.scala          |  3 ---
 core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala | 11 +++++++++++
 core/src/test/scala/unit/kafka/server/KafkaServerTest.scala |  1 -
 4 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index fd48d6cab2a..5ab2456aa4e 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -846,6 +846,9 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
         throw new ConfigException(s"Missing required configuration `${ZkConfigs.ZK_CONNECT_CONFIG}` which has no default value.")
       }
       if (brokerIdGenerationEnable) {
+        if (migrationEnabled) {
+          require(brokerId != -1, "broker id generation is incompatible with migration to ZK. Please disable it before enabling migration")
+        }
         require(brokerId >= -1 && brokerId <= maxReservedBrokerId, "broker.id must be greater than or equal to -1 and not greater than reserved.broker.max.id")
       } else {
         require(brokerId >= 0, "broker.id must be greater than or equal to 0")
@@ -963,6 +966,8 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
     } else {
       // ZK-based
       if (migrationEnabled) {
+        require(brokerId >= 0,
+          "broker broker.id.generation.enable is incompatible with migration to ZK. Please disable it before enabling migration")
         validateQuorumVotersAndQuorumBootstrapServerForMigration()
         require(controllerListenerNames.nonEmpty,
           s"${KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG} must not be empty when running in ZooKeeper migration mode: ${controllerListenerNames.asJava}")
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index e9bc3affcc7..0a9edc32c6b 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -260,9 +260,6 @@ class KafkaServer(
 
         /* generate brokerId */
         config._brokerId = getOrGenerateBrokerId(initialMetaPropsEnsemble)
-        // Currently, we are migrating from ZooKeeper to KRaft. If broker.id.generation.enable is set to true,
-        // we must ensure that the nodeId synchronizes with the broker.id to prevent the nodeId from being -1,
-        // which would result in a failure during the migration.
         config._nodeId = config.brokerId
         logContext = new LogContext(s"[KafkaServer id=${config.brokerId}] ")
         this.logIdent = logContext.logPrefix
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 5e279d17a7b..621b27f21ad 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1883,6 +1883,17 @@ class KafkaConfigTest {
       assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props)).getMessage)
   }
 
+  @Test
+  def testMigrationCannotBeEnabledWithBrokerIdGeneration(): Unit = {
+    val props = TestUtils.createBrokerConfig(-1, TestUtils.MockZkConnect, port = TestUtils.MockZkPort, logDirCount = 2)
+    props.setProperty(KRaftConfigs.MIGRATION_ENABLED_CONFIG, "true")
+    props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "3000@localhost:9093")
+    props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")
+    assertEquals(
+      "requirement failed: broker id generation is incompatible with migration to ZK. Please disable it before enabling migration",
+      assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props)).getMessage)
+  }
+
   @Test
   def testMigrationEnabledKRaftMode(): Unit = {
     val props = new Properties()
diff --git a/core/src/test/scala/unit/kafka/server/KafkaServerTest.scala b/core/src/test/scala/unit/kafka/server/KafkaServerTest.scala
index e8cf3ecfa5c..ea9256baa5e 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaServerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaServerTest.scala
@@ -201,5 +201,4 @@ class KafkaServerTest extends QuorumTestHarness {
     val kafkaConfig = KafkaConfig.fromProps(props)
     TestUtils.createServer(kafkaConfig)
   }
-
 }