Page History

...

• Enable forwarding on all brokers (KIP-590: Redirect Zookeeper Mutation Protocols to The Controller)
• Usage of new BrokerRegistration RPC version
• Usage of new controller RPC versions
• Usage of new ApiVersions RPC version (by KRaft controller only)
• Usage of new ZkMigrationRecordZkMigrationStateRecord
• Enable the migration components on KRaft controller and special migration behavior on ZK brokers

...

For the three ZK controller RPCs UpdateMetadataRequest, LeaderAndIsrRequest, and StopReplicaRequest a new KRaftControllerId IsKRaftController field will be added. This field will point is used to indicate that the active KRaft controller and will only be set when the controller is in KRaft mode. If this field is set, the ControllerId field should be -1. controller sending this RPC is a KRaft controller.

{
  "apiKey": 4,
  "type": "request",
  "listeners": ["zkBroker"],
  "name": "LeaderAndIsrRequest",
  "validVersions": "0-7",  // <-- New version 7
  "flexibleVersions": "4+",
  "fields": [
    { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
      "about": "The controller id." },     
    { "name": "KRaftControllerIdisKRaftController", "type": "int32bool", "versions": "7+", "entityTypedefault": "brokerIdfalse",
      "about": "TheIf KRaft controller id, is used during migration. See KIP-866" }, <-- New field
    { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
      "about": "The controller epoch." },
    ...
   ]
}

...

{
  "apiKey": 5,
  "type": "request",
  "listeners": ["zkBroker"],
  "name": "StopReplicaRequest",
  "validVersions": "0-4",  // <-- New version 4
  "flexibleVersions": "2+",
  "fields": [
    { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
      "about": "The controller id." },     
    { "name": "KRaftControllerIdisKRaftController", "type": "int32bool", "versions": "4+", "entityTypedefault": "brokerIdfalse",
      "about": "TheIf KRaft controller id, is used during migration. See KIP-866" }, // <-- New field
    { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
      "about": "The controller epoch." },
    ...
   ]
}

...

{
  "apiKey": 6,
  "type": "request",
  "listeners": ["zkBroker"],
  "name": "UpdateMetadataRequest",
  "validVersions": "0-8",  // <-- New version 8
  "flexibleVersions": "6+",
  "fields": [
    { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
      "about": "The controller id." },     
    { "name": "KRaftControllerIdisKRaftController", "type": "int32bool", "versions": "8+", "entityTypedefault": "brokerIdfalse",
      "about": "TheIf KRaft controller id, is used during migration. See KIP-866" }, // <-- New field
    { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
      "about": "The controller epoch." },
    ...
   ]
}

...

The active ZooKeeper controller always reports "MigrationIneligible" while the will not report this metric, only the active KRaft controller reports the state corresponding to the state of the migration.

...

The metadata copied from ZK will be encapsulated in a single metadata transaction (KIP-868). A ZkMigrationRecord ZkMigrationStateRecord will also be included in this transaction. 

...

Once the operator has decided to commit to KRaft mode, the final step is to restart the controller quorum and take it out of migration mode by setting zookeeper.metadata.migration.enable to "false" (or unsetting it). The active controller will only finalize the migration once it detects that all members of the quorum have signaled that they are finalizing the migration (again, using the tagged field in ApiVersionsResponse). Once the controller leaves migration mode, it will write a ZkMigrationRecord ZkMigrationStateRecord to the log and no longer perform writes to ZK. It will also disable its special handling of ZK RPCs.

...

UpdateMetadata: for metadata changes, the KRaft controller will need to send UpdateMetadataRequests to the ZK brokers. Instead of ControllerId, the KRaft controller will specify itself using KRaftControllerId field.

StopReplicas: following reassignments and topic deletions, we will need to send StopReplicas to ZK brokers for them to stop managing certain replicas. 

Each of these RPCs will include a new KRaftControllerId IsKRaftController field that points to the active KRaft controller. When this field is present, it acts as a signal to the brokers that the controller is in KRaft modeindicates if the sending controller is a KRaft controller. Using this field, and the zookeeper.metadata.migration.enable config, the brokers can enable migration specific behavior. 

...

A new version of the BrokerRegistration RPC will be used by the ZK brokers to register themselves with KRaft. The ZK brokers will set the new IsMigrationZkBroker field and populate the Features field with a "metadata.version" min and max supported equal to their IBP. The KRaft controller will only accept the registration if the given "metadata.version" is equal to the IBP/MetadataVersion of the quorum. The controller will also only accept the registration if the ZkMigrationReady has a valid value. 

After successfully registering, the ZK brokers will send BrokerHeartbeat RPCs to indicate liveness. The ZK brokers will learn about other brokers in the usual way through UpdateMetadataRequest.

...

For inter-broker requests such as AlterPartitions and ControlledShutdown, we do not want to add the overhead of forwarding so we'll want to include the actual controller in the UpdateMetadataRequest. However, we cannot simply include the KRaft controller as the ControllerId. The ZK brokers connect to a ZK controller by using the "inter.broker.listener.name" config and the node information from LiveBrokers in the UpdateMetadataRequest. For connecting to a KRaft controller, the ZK brokers will need to use the "controller.listener.names" and "controller.quorum.voters" configs. To allow this, we will use the new KRaftControllerId field IsKRaftController field in UpdateMetadataRequest to indicate different controller types to the channel managers.

Topic Deletions

The ZK migration logic will need to deal with asynchronous topic deletions when migrating data. Normally, the ZK controller will complete these asynchronous deletions via TopicDeletionManager. If the KRaft controller takes over before a deletion has occurred, we will need to complete the deletion as part of the ZK to KRaft state migration. Once the migration is complete, we will need to finalize the deletion in ZK so that the state is consistent.

...

Since these two versions contain the same data, but with different field names, we can simply support v0 and v1 in KRaft brokers and avoid modifying the file on disk. By leaving this file unchanged, we better facilitate a downgrade to ZK during the migration. Once the controller has completed the migration and written the final ZkMigrationRecordZkMigrationStateRecord, the brokers can rewrite their meta.properties files as v1 in their log directories.

...

If a migration has been started, but a KRaft controller is elected that is misconfigured (does not have zookeeper.metadata.migration.enable or ZK configs) this controller should resign. When replaying the metadata log during its initialization phase, this controller can see that a migration is in progress by seeing the initial ZkMigrationRecordZkMigrationStateRecord. Since it does not have the required configs, it can resign leadership and throw an error.

If a migration has been finalized, but the KRaft quroum comes up with zookeeper.metadata.migration.enable, we must not re-enter the migration mode. In this case, while replaying the log, the controller can see the second ZkMigrationRecord ZkMigrationStateRecord and know that the migration is finalized and should not be resumed. This should result in errors being thrown, but the quorum can continue operating as normal.

...