Merge pull request #289 from twmb/kip-866

generate/kmsg: support KIP-866 protocols
twmb · Feb 8, 2023 · ede1adc · ede1adc
2 parents a4cd5ce + 155c9c6
commit ede1adc
Show file tree

Hide file tree

Showing 7 changed files with 100 additions and 13 deletions.
diff --git a/generate/definitions/04_leader_and_isr b/generate/definitions/04_leader_and_isr
@@ -31,8 +31,11 @@ LeaderAndISRResponseTopicPartition => not top level, no encoding, flexible v4+
 // Kafka 1.0 introduced version 1. Kafka 2.2 introduced version 2, proposed
 // in KIP-380, which changed the layout of the struct to be more memory
 // efficient. Kafka 2.4.0 introduced version 3 with KIP-455.
-LeaderAndISRRequest => key 4, max version 6, flexible v4+
+// Kafka 3.4 introduced version 7 with KIP-866.
+LeaderAndISRRequest => key 4, max version 7, flexible v4+
   ControllerID: int32
+  // If KRaft controller id is used during migration. See KIP-866.
+  IsKRaftController: bool // v7+
   ControllerEpoch: int32
   BrokerEpoch: int64(-1) // v2+
   Type: int8 // v5+

diff --git a/generate/definitions/05_stop_replica b/generate/definitions/05_stop_replica
@@ -8,9 +8,12 @@
 //
 // Kafka 2.6 introduced version 3, proposed in KIP-570, reorganizes partitions
 // to be stored and adds the leader epoch and delete partition fields per partition.
-StopReplicaRequest => key 5, max version 3, flexible v2+
+// Kafka 3.4 introduced version 4 with KIP-866.
+StopReplicaRequest => key 5, max version 4, flexible v2+
   ControllerID: int32
   ControllerEpoch: int32
+  // If KRaft controller id is used during migration. See KIP-866.
+  IsKRaftController: bool // v4+
   BrokerEpoch: int64(-1) // v1+
   DeletePartitions: bool // v0-v2
   Topics: [=>]

diff --git a/generate/definitions/06_update_metadata b/generate/definitions/06_update_metadata
@@ -19,8 +19,11 @@ UpdateMetadataRequestTopicPartition => not top level, no encoding, flexible v6+
 //
 // Kafka 2.2 introduced version 5, proposed in KIP-380, which changed the
 // layout of the struct to be more memory efficient.
-UpdateMetadataRequest => key 6, max version 7, flexible v6+
+// Kafka 3.4 introduced version 8 with KIP-866.
+UpdateMetadataRequest => key 6, max version 8, flexible v6+
   ControllerID: int32
+  // If KRaft controller id is used during migration. See KIP-866.
+  IsKRaftController: bool // v8+
   ControllerEpoch: int32
   BrokerEpoch: int64(-1) // v5+
   PartitionStates: [UpdateMetadataRequestTopicPartition] // v0-v4

diff --git a/generate/definitions/18_api_versions b/generate/definitions/18_api_versions
@@ -65,3 +65,6 @@ ApiVersionsResponse =>
     MaxVersionLevel: int16
     // The cluster-wide finalized min version level for the feature.
     MinVersionLevel: int16
+  // Set by a KRaft controller if the required configurations for ZK migration
+  // are present
+  ZkMigrationReady: bool // tag 3
diff --git a/generate/definitions/62_broker_registration b/generate/definitions/62_broker_registration
@@ -1,6 +1,6 @@
 // For KIP-500 / KIP-631, BrokerRegistrationRequest is an internal
 // broker-to-broker only request.
-BrokerRegistrationRequest => key 62, max version 0, flexible v0+
+BrokerRegistrationRequest => key 62, max version 1, flexible v0+
   // The broker ID.
   BrokerID: int32
   // The cluster ID of the broker process.
@@ -27,6 +27,9 @@ BrokerRegistrationRequest => key 62, max version 0, flexible v0+
     MaxSupportedVersion: int16
   // The rack that this broker is in, if any.
   Rack: nullable-string
+  // If the required configurations for ZK migration are present, this value is
+  // set to true.
+  IsMigratingZkBroker: bool // v1+
 
 // BrokerRegistrationResponse is a response to a BrokerRegistrationRequest.
 BrokerRegistrationResponse =>

diff --git a/pkg/kmsg/generated.go b/pkg/kmsg/generated.go
diff --git a/pkg/kversion/kversion.go b/pkg/kversion/kversion.go
@@ -788,7 +788,7 @@ var max270 = nextMax(max260, func(v listenerKeys) listenerKeys {
 var max280 = nextMax(max270, func(v listenerKeys) listenerKeys {
 	// KAFKA-10181 KAFKA-10181 KIP-590
 	v = append(v,
-		k(rController), // 58 envelope
+		k(zkBroker, rController), // 58 envelope, controller first, zk in KAFKA-14446 8b045dcbf6b89e1a9594ff95642d4882765e4b0d KIP-866 Kafka 3.4
 	)
 
 	// KAFKA-10729 85f94d50271c952c3e9ee49c4fc814c0da411618 KIP-482
@@ -939,6 +939,9 @@ var (
 	maxTip    = nextMax(maxStable, func(v listenerKeys) listenerKeys {
 		// KAFKA-14304 7b7e40a536a79cebf35cc278b9375c8352d342b9 KIP-866
 		// KAFKA-14448 67c72596afe58363eceeb32084c5c04637a33831 added BrokerRegistration
+		// KAFKA-14493 db490707606855c265bc938e1b236070e0e2eba5 changed BrokerRegistration
+		// KAFKA-14304 0bb05d8679b684ad8fbb2eb40dfc00066186a75a changed BrokerRegistration back to a bool...
+		// 5b521031edea8ea7cbcca7dc24a58429423740ff added tag to ApiVersions
 		v[4].inc()  // 7 leader and isr
 		v[5].inc()  // 4 stop replica
 		v[6].inc()  // 8 update metadata