Merge pull request #500 from twmb/next-minor

Next minor

README.md

@@ -340,6 +340,7 @@ generation.
 | [KIP-392](https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica) — Closest replica fetching w/ rack | 2.2 | Supported |
 | [KIP-394](https://cwiki.apache.org/confluence/display/KAFKA/KIP-394%3A+Require+member.id+for+initial+join+group+request) — Require member.id for initial join request |  2.2 | Supported |
 | [KIP-396](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=97551484) — Commit offsets manually | 2.4 | Supported |
+| [KIP-405](https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage) — Kafka Tiered Storage | 3.5 | Supported (protos) |
 | [KIP-412](https://cwiki.apache.org/confluence/display/KAFKA/KIP-412%3A+Extend+Admin+API+to+support+dynamic+application+log+levels) — Dynamic log levels w/ IncrementalAlterConfigs | 2.4 | Supported |
 | [KIP-429](https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol) — Incremental rebalance (see KAFKA-8179) | 2.4 | Supported |
 | [KIP-430](https://cwiki.apache.org/confluence/display/KAFKA/KIP-430+-+Return+Authorized+Operations+in+Describe+Responses) — Authorized ops in DescribeGroups | 2.3 | Supported |
@@ -397,6 +398,7 @@ generation.
 | [KIP-866](https://cwiki.apache.org/confluence/display/KAFKA/KIP-866+ZooKeeper+to+KRaft+Migration) — ZK to Raft RPC changes | 3.4 | Supported |
 | [KIP-893](https://cwiki.apache.org/confluence/display/KAFKA/KIP-893%3A+The+Kafka+protocol+should+support+nullable+structs) — Nullable structs in the protocol | 3.5 | Supported |
 | [KIP-899](https://cwiki.apache.org/confluence/display/KAFKA/KIP-899%3A+Allow+clients+to+rebootstrap) — Allow clients to rebootstrap | ? | Supported (`UpdateSeedBrokers`) |
+| [KIP-903](https://cwiki.apache.org/confluence/display/KAFKA/KIP-903%3A+Replicas+with+stale+broker+epoch+should+not+be+allowed+to+join+the+ISR) — Stale broker epoch fencing| 3.5 | Supported (proto) |
 
 Missing from above but included in librdkafka is:
 

generate/definitions/01_fetch

@@ -10,14 +10,23 @@
 //
 // Starting in v13, topics must use UUIDs rather than their string name
 // identifiers.
-FetchRequest => key 1, max version 13, flexible v12+
+//
+// Version 15 adds the ReplicaState which includes new field ReplicaEpoch and
+// the ReplicaID, and deprecates the old ReplicaID (KIP-903).
+FetchRequest => key 1, max version 15, flexible v12+
   // The cluster ID, if known. This is used to validate metadata fetches
   // prior to broker registration.
   ClusterID: nullable-string(null) // tag 0
   // ReplicaID is the broker ID of performing the fetch request. Standard
   // clients should use -1. To be a "debug" replica, use -2. The debug
   // replica can be used to fetch messages from non-leaders.
-  ReplicaID: int32
+  ReplicaID: int32(-1) // v0-v14
+  // ReplicaState is a broker-only tag for v15+, see KIP-903 for more details.
+  ReplicaState: => // tag 1
+    // The replica ID of the follower, or -1 if this request is from a consumer.
+    ID: int32(-1)
+    // The epoch of this follower, or -1 if not available.
+    Epoch: int64(-1)
   // MaxWaitMillis is how long to wait for MinBytes to be hit before a broker
   // responds to a fetch request.
   MaxWaitMillis: int32
@@ -158,7 +167,11 @@ FetchResponse =>
       // OFFSET_OUT_OF_RANGE is returned if requesting an offset past the
       // current end offset or before the beginning offset.
       //
-      // UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown.
+      // UNKNOWN_TOPIC_ID is returned if using uuid's and the uuid is unknown
+      // (v13+ / Kafka 3.1+).
+      //
+      // OFFSET_MOVED_TO_TIERED_STORAGE is returned if a follower is trying to
+      // fetch from an offset that is now in tiered storage.
       ErrorCode: int16
       // HighWatermark is the current high watermark for this partition,
       // that is, the current offset that is on all in sync replicas.

generate/definitions/02_list_offsets

@@ -7,7 +7,10 @@
 //
 // Version 7, introduced in Kafka 3.0, supports -3 as a timestamp to return
 // the timestamp and offset for the record with the largest timestamp.
-ListOffsetsRequest => key 2, max version 7, flexible v6+
+//
+// Version 8, introduced in Kafka 3.4, supports -4 as a timestamp to return
+// the local log start offset (in the context of tiered storage, see KIP-405).
+ListOffsetsRequest => key 2, max version 8, flexible v6+
   // ReplicaID is the broker ID to get offsets from. As a Kafka client, use -1.
   // The consumer replica ID (-1) causes requests to only succeed if issued
   // against the leader broker.
@@ -53,6 +56,11 @@ ListOffsetsRequest => key 2, max version 7, flexible v6+
       // to determine topic "liveness" (when was the last produce?).
       // Previously, this was not easy to determine. See KIP-734 for more
       // detail.
+      //
+      // If you are talking to Kafka 3.4+ and using request version 8+ (for
+      // KIP-405), the new special timestamp -4 returns the local log start
+      // offset. In the context of tiered storage, the earliest local log start
+      // offset is the offset actually available on disk on the broker.
       Timestamp: int64
       // MaxNumOffsets is the maximum number of offsets to report.
       // This was removed after v0.

generate/definitions/56_alter_partition

@@ -1,6 +1,8 @@
 // AlterPartitionRequest, proposed in KIP-497 and introduced in Kafka 2.7.0,
 // is an admin request to modify ISR.
-AlterPartitionRequest => key 56, max version 2, flexible v0+, admin
+//
+// Version 3 was added for KIP-903 and replaced NewISR.
+AlterPartitionRequest => key 56, max version 3, flexible v0+, admin
   // The ID of the requesting broker.
   BrokerID: int32
   // The epoch of the requesting broker.
@@ -13,7 +15,12 @@ AlterPartitionRequest => key 56, max version 2, flexible v0+, admin
       // The leader epoch of this partition.
       LeaderEpoch: int32
       // The ISR for this partition.
-      NewISR: [int32]
+      NewISR: [int32] // v0-v2
+      NewEpochISR: [=>] // v3+
+        // The broker ID .
+        BrokerID: int32
+        // The broker's epoch; -1 if the epoch check is not supported.
+        BrokerEpoch: int32(-1)
       // 1 if the partition is recovering from unclean leader election; 0 otherwise
       LeaderRecoveryState: int8 // v1+
       // The expected epoch of the partition which is being updated.

pkg/kerr/kerr.go

@@ -182,6 +182,7 @@ var (
 	FetchSessionTopicIDError           = &Error{"FETCH_SESSION_TOPIC_ID_ERROR", 106, true, "The fetch session encountered inconsistent topic ID usage."}
 	IneligibleReplica                  = &Error{"INELIGIBLE_REPLICA", 107, false, "The new ISR contains at least one ineligible replica."}
 	NewLeaderElected                   = &Error{"NEW_LEADER_ELECTED", 108, false, "The AlterPartition request successfully updated the partition state but the leader has changed."}
+	OffsetMovedToTieredStorage         = &Error{"OFFSET_MOVED_TO_TIERED_STORAGE", 109, false, "The requested offset is moved to tiered storage."}
 )
 
 var code2err = map[int16]error{
@@ -295,4 +296,5 @@ var code2err = map[int16]error{
 	106: FetchSessionTopicIDError,
 	107: IneligibleReplica,
 	108: NewLeaderElected,
+	109: OffsetMovedToTieredStorage,
 }