KAFKA-17543: Enforce that broker.id.generation.enable is not used whe…

…n migrating to KRaft (#17192) Reviewers: Chia-Ping Tsai <[email protected]>, David Arthur <[email protected]>
apache · Sep 14, 2024 · d7a456e · d7a456e
1 parent 6610a4d
commit d7a456e
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 4 deletions.
diff --git 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
@@ -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
@@ -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
@@ -201,5 +201,4 @@ class KafkaServerTest extends QuorumTestHarness {
     val kafkaConfig = KafkaConfig.fromProps(props)
     TestUtils.createServer(kafkaConfig)
   }
-
 }
-Original file line number
+Diff line change
@@ Expand Up / @@ -201,5 +201,4 @@ class KafkaServerTest extends QuorumTestHarness { @@
         val kafkaConfig = KafkaConfig.fromProps(props)
         TestUtils.createServer(kafkaConfig)
       }
     }