KAFKA-14805 KRaft controller supports pre-migration mode

checkstyle/suppressions.xml

@@ -306,9 +306,9 @@
     <suppress checks="(ParameterNumber|ClassDataAbstractionCoupling)"
               files="(QuorumController).java"/>
     <suppress checks="CyclomaticComplexity"
-              files="(ClientQuotasImage|KafkaEventQueue|MetadataDelta|QuorumController|ReplicationControlManager).java"/>
+              files="(ClientQuotasImage|KafkaEventQueue|MetadataDelta|QuorumController|ReplicationControlManager|KRaftMigrationDriver|ClusterControlManager).java"/>
     <suppress checks="NPathComplexity"
-              files="(ClientQuotasImage|KafkaEventQueue|ReplicationControlManager|FeatureControlManager|KRaftMigrationDriver|ScramControlManager).java"/>
+              files="(ClientQuotasImage|KafkaEventQueue|ReplicationControlManager|FeatureControlManager|KRaftMigrationDriver|ScramControlManager|ClusterControlManager).java"/>
     <suppress checks="(NPathComplexity|ClassFanOutComplexity|CyclomaticComplexity|ClassDataAbstractionCoupling|LocalVariableName|MemberName|ParameterName|MethodLength|JavaNCSS|AvoidStarImport)"
             files="metadata[\\/]src[\\/](generated|generated-test)[\\/].+.java$"/>
     <suppress checks="BooleanExpressionComplexity"

core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java

@@ -105,19 +105,20 @@ public List<Extension> getAdditionalExtensions() {
                     zkReference.set(new EmbeddedZookeeper());
                     builder.setConfigProp("zookeeper.connect", String.format("localhost:%d", zkReference.get().port()));
                 }
-
                 // Copy properties into the TestKit builder
                 clusterConfig.serverProperties().forEach((key, value) -> builder.setConfigProp(key.toString(), value.toString()));
                 // KAFKA-12512 need to pass security protocol and listener name here
                 KafkaClusterTestKit cluster = builder.build();
                 clusterReference.set(cluster);
                 cluster.format();
-                cluster.startup();
-                kafka.utils.TestUtils.waitUntilTrue(
-                    () -> cluster.brokers().get(0).brokerState() == BrokerState.RUNNING,
-                    () -> "Broker never made it to RUNNING state.",
-                    org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
-                    100L);
+                if (clusterConfig.isAutoStart()) {
+                    cluster.startup();
+                    kafka.utils.TestUtils.waitUntilTrue(
+                        () -> cluster.brokers().get(0).brokerState() == BrokerState.RUNNING,
+                        () -> "Broker never made it to RUNNING state.",
+                        org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
+                        100L);
+                }
             },
             (AfterTestExecutionCallback) context -> clusterInstance.stop(),
             new ClusterInstanceParameterResolver(clusterInstance),

core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala

@@ -44,7 +44,7 @@ import org.apache.kafka.metadata.migration.ZkMigrationLeadershipState
 import org.apache.kafka.raft.RaftConfig
 import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion, ProducerIdsBlock}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertTrue}
-import org.junit.jupiter.api.{Disabled, Timeout}
+import org.junit.jupiter.api.Timeout
 import org.junit.jupiter.api.extension.ExtendWith
 import org.slf4j.LoggerFactory
 
@@ -120,6 +120,11 @@ class ZkMigrationIntegrationTest {
     }
   }
 
+  /**
+   * Test ZkMigrationClient against a real ZooKeeper-backed Kafka cluster. This test creates a ZK cluster
+   * and modifies data using AdminClient. The ZkMigrationClient is then used to read the metadata from ZK
+   * as would happen during a migration. The generated records are then verified.
+   */
   @ClusterTest(brokers = 3, clusterType = Type.ZK, metadataVersion = MetadataVersion.IBP_3_4_IV0)
   def testMigrate(clusterInstance: ClusterInstance): Unit = {
     val admin = clusterInstance.createAdminClient()
@@ -182,7 +187,6 @@ class ZkMigrationIntegrationTest {
     migrationState = migrationClient.releaseControllerLeadership(migrationState)
   }
 
-  @Disabled("Will be fixed by KAFKA-14840")
   @ClusterTest(clusterType = Type.ZK, brokers = 3, metadataVersion = MetadataVersion.IBP_3_4_IV0, serverProperties = Array(
     new ClusterConfigProperty(key = "inter.broker.listener.name", value = "EXTERNAL"),
     new ClusterConfigProperty(key = "listeners", value = "PLAINTEXT://localhost:0,EXTERNAL://localhost:0"),

core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala

@@ -15,23 +15,24 @@
  * limitations under the License.
  */
 
-package unit.kafka.server
+package kafka.server
 
-import kafka.server.{BrokerToControllerChannelManager, ControllerInformation, ControllerNodeProvider, ControllerRequestCompletionHandler}
 import kafka.test.ClusterInstance
-import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, Type}
+import kafka.test.annotation._
 import kafka.test.junit.ClusterTestExtensions
+import kafka.test.junit.RaftClusterInvocationContext.RaftClusterInstance
 import org.apache.kafka.clients.ClientResponse
-import org.apache.kafka.common.message.{BrokerRegistrationRequestData, BrokerRegistrationResponseData}
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic
+import org.apache.kafka.common.message.{BrokerRegistrationRequestData, CreateTopicsRequestData}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.{BrokerRegistrationRequest, BrokerRegistrationResponse}
+import org.apache.kafka.common.requests._
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.Time
 import org.apache.kafka.common.{Node, Uuid}
 import org.apache.kafka.server.common.MetadataVersion
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
 import org.junit.jupiter.api.extension.ExtendWith
 import org.junit.jupiter.api.{Tag, Timeout}
 
@@ -74,18 +75,19 @@ class BrokerRegistrationRequestTest {
     )
   }
 
-  def sendAndRecieve(
+  def sendAndReceive[T <: AbstractRequest, R <: AbstractResponse](
     channelManager: BrokerToControllerChannelManager,
-    req: BrokerRegistrationRequestData
-  ): BrokerRegistrationResponseData = {
-    val responseFuture = new CompletableFuture[BrokerRegistrationResponseData]()
-    channelManager.sendRequest(new BrokerRegistrationRequest.Builder(req), new ControllerRequestCompletionHandler() {
+    reqBuilder: AbstractRequest.Builder[T],
+    timeoutMs: Int
+  ): R = {
+    val responseFuture = new CompletableFuture[R]()
+    channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
       override def onTimeout(): Unit = responseFuture.completeExceptionally(new TimeoutException())
 
       override def onComplete(response: ClientResponse): Unit =
-        responseFuture.complete(response.responseBody().asInstanceOf[BrokerRegistrationResponse].data())
+        responseFuture.complete(response.responseBody().asInstanceOf[R])
     })
-    responseFuture.get(30, TimeUnit.SECONDS)
+    responseFuture.get(timeoutMs, TimeUnit.MILLISECONDS)
   }
 
   def registerBroker(
@@ -113,7 +115,22 @@ class BrokerRegistrationRequestTest {
       .setIsMigratingZkBroker(zkEpoch.isDefined)
       .setFeatures(features)
 
-    Errors.forCode(sendAndRecieve(channelManager, req).errorCode())
+    val resp = sendAndReceive[BrokerRegistrationRequest, BrokerRegistrationResponse](
+      channelManager, new BrokerRegistrationRequest.Builder(req), 30000)
+    Errors.forCode(resp.data().errorCode())
+  }
+
+
+  def createTopics(channelManager: BrokerToControllerChannelManager,
+                   topicName: String): Errors = {
+    val createTopics = new CreateTopicsRequestData()
+    createTopics.setTopics(new CreateTopicsRequestData.CreatableTopicCollection())
+    createTopics.topics().add(new CreatableTopic().setName(topicName).setNumPartitions(10).setReplicationFactor(1))
+    createTopics.setTimeoutMs(500)
+
+    val req = new CreateTopicsRequest.Builder(createTopics)
+    val resp = sendAndReceive[CreateTopicsRequest, CreateTopicsResponse](channelManager, req, 3000).data()
+    Errors.forCode(resp.topics().find(topicName).errorCode())
   }
 
   @ClusterTest(clusterType = Type.KRAFT, brokers = 0, controllers = 1, metadataVersion = MetadataVersion.IBP_3_4_IV0,
@@ -145,25 +162,29 @@ class BrokerRegistrationRequestTest {
   }
 
   @ClusterTest(clusterType = Type.KRAFT, brokers = 0, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3,
-    serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "true")))
-  def testRegisterZkWithKRaftOldMetadataVersion(clusterInstance: ClusterInstance): Unit = {
+    serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "false")))
+  def testRegisterZkWith33Controller(clusterInstance: ClusterInstance): Unit = {
+    // Verify that a controller running an old metadata.version cannot register a ZK broker
     val clusterId = clusterInstance.clusterId()
     val channelManager = brokerToControllerChannelManager(clusterInstance)
     try {
       channelManager.start()
-
+      // Invalid registration (isMigratingZkBroker, but MV does not support migrations)
       assertEquals(
         Errors.BROKER_ID_NOT_REGISTERED,
-        registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_3_IV0, MetadataVersion.IBP_3_3_IV0))))
+        registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_3_IV0, MetadataVersion.IBP_3_3_IV3))))
 
+      // No features (MV) sent with registration, controller can't verify
       assertEquals(
         Errors.BROKER_ID_NOT_REGISTERED,
         registerBroker(channelManager, clusterId, 100, Some(1), None))
 
+      // Given MV is too high for controller to support
       assertEquals(
         Errors.BROKER_ID_NOT_REGISTERED,
         registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_4_IV0, MetadataVersion.IBP_3_4_IV0))))
 
+      // Controller supports this MV and isMigratingZkBroker is false, so this one works
       assertEquals(
         Errors.NONE,
         registerBroker(channelManager, clusterId, 100, None, Some((MetadataVersion.IBP_3_3_IV3, MetadataVersion.IBP_3_4_IV0))))
@@ -172,9 +193,16 @@ class BrokerRegistrationRequestTest {
     }
   }
 
-  @ClusterTest(clusterType = Type.KRAFT, brokers = 0, controllers = 1, metadataVersion = MetadataVersion.IBP_3_4_IV0,
+  @ClusterTest(
+    clusterType = Type.KRAFT,
+    brokers = 1,
+    controllers = 1,
+    metadataVersion = MetadataVersion.IBP_3_4_IV0,
+    autoStart = AutoStart.NO,
     serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "true")))
   def testRegisterZkWithKRaftMigrationEnabled(clusterInstance: ClusterInstance): Unit = {
+    clusterInstance.asInstanceOf[RaftClusterInstance].controllers().forEach(_.startup())
+
     val clusterId = clusterInstance.clusterId()
     val channelManager = brokerToControllerChannelManager(clusterInstance)
     try {
@@ -192,11 +220,32 @@ class BrokerRegistrationRequestTest {
         Errors.UNSUPPORTED_VERSION,
         registerBroker(channelManager, clusterId, 100, Some(1), Some((MetadataVersion.IBP_3_3_IV3, MetadataVersion.IBP_3_3_IV3))))
 
+      // Cannot register KRaft broker when in pre-migration
       assertEquals(
-        Errors.NONE,
+        Errors.BROKER_ID_NOT_REGISTERED,
         registerBroker(channelManager, clusterId, 100, None, Some((MetadataVersion.IBP_3_4_IV0, MetadataVersion.IBP_3_4_IV0))))
     } finally {
       channelManager.shutdown()
     }
   }
+
+  /**
+   * Start a KRaft cluster with migrations enabled, verify that the controller does not accept metadata changes
+   * through the RPCs. The migration never proceeds past pre-migration since no ZK brokers are registered.
+   */
+  @ClusterTests(Array(
+    new ClusterTest(clusterType = Type.KRAFT, autoStart = AutoStart.NO, controllers = 1, metadataVersion = MetadataVersion.IBP_3_4_IV0,
+      serverProperties = Array(new ClusterConfigProperty(key = "zookeeper.metadata.migration.enable", value = "true")))
+  ))
+  def testNoMetadataChangesInPreMigrationMode(clusterInstance: ClusterInstance): Unit = {
+    clusterInstance.asInstanceOf[RaftClusterInstance].controllers().forEach(_.startup())
+
+    val channelManager = brokerToControllerChannelManager(clusterInstance)
+    try {
+      channelManager.start()
+      assertThrows(classOf[TimeoutException], () => createTopics(channelManager, "test-pre-migration"))
+    } finally {
+      channelManager.shutdown()
+    }
+  }
 }

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -340,6 +340,11 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
             throw new BrokerIdNotRegisteredException("Controller does not support registering ZK brokers.");
         }
 
+        if (!request.isMigratingZkBroker() && featureControl.inPreMigrationMode()) {
+            throw new BrokerIdNotRegisteredException("Controller is in pre-migration mode and cannot register KRaft " +
+                "brokers until the metadata migration is complete.");
+        }
+
         RegisterBrokerRecord record = new RegisterBrokerRecord().
             setBrokerId(brokerId).
             setIsMigratingZkBroker(request.isMigratingZkBroker()).

metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java

@@ -30,11 +30,13 @@
 import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.admin.FeatureUpdate;
 import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.metadata.ZkMigrationStateRecord;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.ApiError;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.metadata.FinalizedControllerFeatures;
 import org.apache.kafka.metadata.VersionRange;
+import org.apache.kafka.metadata.migration.ZkMigrationState;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.timeline.SnapshotRegistry;
@@ -46,6 +48,7 @@
 
 
 public class FeatureControlManager {
+
     public static class Builder {
         private LogContext logContext = null;
         private SnapshotRegistry snapshotRegistry = null;
@@ -85,11 +88,13 @@ public FeatureControlManager build() {
                 quorumFeatures = new QuorumFeatures(0, new ApiVersions(), QuorumFeatures.defaultFeatureMap(),
                         Collections.emptyList());
             }
-            return new FeatureControlManager(logContext,
+            return new FeatureControlManager(
+                logContext,
                 quorumFeatures,
                 snapshotRegistry,
                 metadataVersion,
-                minimumBootstrapVersion);
+                minimumBootstrapVersion
+            );
         }
     }
 
@@ -110,6 +115,11 @@ public FeatureControlManager build() {
      */
     private final TimelineObject<MetadataVersion> metadataVersion;
 
+    /**
+     * The current ZK migration state
+     */
+    private final TimelineObject<ZkMigrationState> migrationControlState;
+
     /**
      * The minimum bootstrap version that we can't downgrade before.
      */
@@ -127,6 +137,7 @@ private FeatureControlManager(
         this.finalizedVersions = new TimelineHashMap<>(snapshotRegistry, 0);
         this.metadataVersion = new TimelineObject<>(snapshotRegistry, metadataVersion);
         this.minimumBootstrapVersion = minimumBootstrapVersion;
+        this.migrationControlState = new TimelineObject<>(snapshotRegistry, ZkMigrationState.NONE);
     }
 
     ControllerResult<Map<String, ApiError>> updateFeatures(
@@ -153,6 +164,10 @@ MetadataVersion metadataVersion() {
         return metadataVersion.get();
     }
 
+    ZkMigrationState zkMigrationState() {
+        return migrationControlState.get();
+    }
+
     private ApiError updateFeature(
         String featureName,
         short newVersion,
@@ -232,13 +247,20 @@ private ApiError updateMetadataVersion(
         Consumer<ApiMessageAndVersion> recordConsumer
     ) {
         MetadataVersion currentVersion = metadataVersion();
+        ZkMigrationState zkMigrationState = zkMigrationState();
         final MetadataVersion newVersion;
         try {
             newVersion = MetadataVersion.fromFeatureLevel(newVersionLevel);
         } catch (IllegalArgumentException e) {
             return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
         }
 
+        // Don't allow metadata.version changes while we're migrating
+        if (zkMigrationState.inProgress()) {
+            return invalidMetadataVersion(newVersionLevel, "Unable to modify metadata.version while a " +
+                "ZK migration is in progress.");
+        }
+
         // We cannot set a version earlier than IBP_3_3_IV0, since that was the first version that contained
         // FeatureLevelRecord itself.
         if (newVersion.isLessThan(minimumBootstrapVersion)) {
@@ -266,6 +288,7 @@ private ApiError updateMetadataVersion(
             new FeatureLevelRecord()
                 .setName(MetadataVersion.FEATURE_NAME)
                 .setFeatureLevel(newVersionLevel), FEATURE_LEVEL_RECORD.lowestSupportedVersion()));
+
         return ApiError.NONE;
     }
 
@@ -284,6 +307,16 @@ FinalizedControllerFeatures finalizedFeatures(long epoch) {
         return new FinalizedControllerFeatures(features, epoch);
     }
 
+    /**
+     * Tests if the controller should be preventing metadata updates due to being in the PRE_MIGRATION
+     * state. If the controller does not yet support migrations (before 3.4-IV0), then the migration state
+     * will be NONE and this will return false. Once the controller has been upgraded to a version that supports
+     * migrations, then this method checks if the migration state is equal to PRE_MIGRATION.
+     */
+    boolean inPreMigrationMode() {
+        return migrationControlState.get().equals(ZkMigrationState.PRE_MIGRATION);
+    }
+
     public void replay(FeatureLevelRecord record) {
         VersionRange range = quorumFeatures.localSupportedFeature(record.name());
         if (!range.contains(record.featureLevel())) {
@@ -304,6 +337,13 @@ public void replay(FeatureLevelRecord record) {
         }
     }
 
+    public void replay(ZkMigrationStateRecord record) {
+        ZkMigrationState recordState = ZkMigrationState.of(record.zkMigrationState());
+        ZkMigrationState currentState = migrationControlState.get();
+        log.info("Transitioning ZK migration state from {} to {}", currentState, recordState);
+        migrationControlState.set(recordState);
+    }
+
     boolean isControllerId(int nodeId) {
         return quorumFeatures.isControllerId(nodeId);
     }