impl protocol and server side

TaiJuWu · TaiJuWu · commit 7fc04884bfe5 · 2025-12-10T10:52:23.000+08:00
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopicCollection;
@@ -70,6 +71,24 @@ public OffsetsForLeaderEpochRequest build(short version) {
             if (version < oldestAllowedVersion() || version > latestAllowedVersion())
                 throw new UnsupportedVersionException("Cannot build " + this + " with version " + version);
 
+            if (version <= 4) {
+                for (OffsetForLeaderEpochRequestData.OffsetForLeaderTopic topic : data.topics()) {
+                    if (topic.topic() == null || topic.topic().isEmpty()) {
+                        throw new UnsupportedVersionException("The broker offsets for leader api version " +
+                                version + " does require usage of topic names.");
+                    }
+                }
+            }
+
+            if (version >= 5) {
+                for (OffsetForLeaderEpochRequestData.OffsetForLeaderTopic topic : data.topics()) {
+                    if (topic.topicId() == null || topic.topicId() == Uuid.ZERO_UUID) {
+                        throw new UnsupportedVersionException("The broker offsets for leader api version " +
+                                version + " does require usage of topic ids.");
+                    }
+                }
+            }
+
             return new OffsetsForLeaderEpochRequest(data, version);
         }
 
@@ -93,6 +112,10 @@ public int replicaId() {
         return data.replicaId();
     }
 
+    public static boolean useTopicIds(short version) {
+        return version >= 5;
+    }
+
     public static OffsetsForLeaderEpochRequest parse(Readable readable, short version) {
         return new OffsetsForLeaderEpochRequest(new OffsetForLeaderEpochRequestData(readable, version), version);
     }
@@ -104,7 +127,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         OffsetForLeaderEpochResponseData responseData = new OffsetForLeaderEpochResponseData();
         data.topics().forEach(topic -> {
             OffsetForLeaderTopicResult topicData = new OffsetForLeaderTopicResult()
-                .setTopic(topic.topic());
+                .setTopic(topic.topic()).setTopicId(topic.topicId());
             topic.partitions().forEach(partition ->
                 topicData.partitions().add(new EpochEndOffset()
                     .setPartition(partition.partition())
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochResponse.java
@@ -63,6 +63,10 @@ public Map<Errors, Integer> errorCounts() {
         return errorCounts;
     }
 
+    public static boolean useTopicIds(short version) {
+        return version >= 5;
+    }
+
     public int throttleTimeMs() {
         return data.throttleTimeMs();
     }
diff --git a/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json b/clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json
@@ -29,15 +29,19 @@
   //    Followers will use this replicaId when using an older version of the protocol.
   //
   // Version 4 enables flexible versions.
-  "validVersions": "2-4",
+  //
+  // Version 5 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "2-5",
   "flexibleVersions": "4+",
   "fields": [
     { "name": "ReplicaId", "type": "int32", "versions": "3+", "default": -2, "ignorable": true, "entityType": "brokerId",
       "about": "The broker ID of the follower, of -1 if this request is from a consumer." },
     { "name": "Topics", "type": "[]OffsetForLeaderTopic", "versions": "0+",
       "about": "Each topic to get offsets for.", "fields": [
-      { "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName",
-        "mapKey": true, "about": "The topic name." },
+      { "name": "Topic", "type": "string", "versions": "0-4", "entityType": "topicName",
+        "mapKey": true, "ignorable": true, "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "5+", "ignorable": true,
+        "about": "The unique topic ID."},
       { "name": "Partitions", "type": "[]OffsetForLeaderPartition", "versions": "0+",
         "about": "Each partition to get offsets for.", "fields": [
         { "name": "Partition", "type": "int32", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/OffsetForLeaderEpochResponse.json b/clients/src/main/resources/common/message/OffsetForLeaderEpochResponse.json
@@ -26,15 +26,19 @@
   // Version 3 is the same as version 2.
   //
   // Version 4 enables flexible versions.
-  "validVersions": "2-4",
+  //
+  // Version 5 replaces topic names with topic IDs (KIP-516). May return UNKNOWN_TOPIC_ID error code.
+  "validVersions": "2-5",
   "flexibleVersions": "4+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "2+", "ignorable": true,
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name": "Topics", "type": "[]OffsetForLeaderTopicResult", "versions": "0+",
       "about": "Each topic we fetched offsets for.", "fields": [
-      { "name": "Topic", "type": "string", "versions": "0+", "entityType": "topicName",
-        "mapKey": true, "about": "The topic name." },
+      { "name": "Topic", "type": "string", "versions": "0-4", "entityType": "topicName",
+        "mapKey": true, "ignorable": true, "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "5+", "ignorable": true,
+        "about": "The unique topic ID."},
       { "name": "Partitions", "type": "[]EpochEndOffset", "versions": "0+",
         "about": "Each partition in the topic we fetched offsets for.", "fields": [
         { "name": "ErrorCode", "type": "int16", "versions": "0+",
diff --git a/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
@@ -378,36 +378,24 @@ public void testThrottleTimeIgnorableInDescribeGroupsResponse() throws Exception
         testAllMessageRoundTripsBeforeVersion((short) 1, responseWithGroupInstanceId, expectedResponse);
     }
 
-    @Test
-    public void testOffsetForLeaderEpochVersions() throws Exception {
-        // Version 2 adds optional current leader epoch
-        OffsetForLeaderEpochRequestData.OffsetForLeaderPartition partitionDataNoCurrentEpoch =
-                new OffsetForLeaderEpochRequestData.OffsetForLeaderPartition()
-                        .setPartition(0)
-                        .setLeaderEpoch(3);
-        OffsetForLeaderEpochRequestData.OffsetForLeaderPartition partitionDataWithCurrentEpoch =
-                new OffsetForLeaderEpochRequestData.OffsetForLeaderPartition()
+    @ParameterizedTest
+    @ApiKeyVersionsSource(apiKey = ApiKeys.OFFSET_FOR_LEADER_EPOCH)
+    public void testOffsetForLeaderEpochVersions(short version) throws Exception {
+        OffsetForLeaderEpochRequestData request = new OffsetForLeaderEpochRequestData()
+            .setReplicaId(version >= 3 ? 5 : -2);
+        request.topics().add(
+            new OffsetForLeaderEpochRequestData.OffsetForLeaderTopic()
+                .setTopicId(version >= 5 ? Uuid.randomUuid() : Uuid.ZERO_UUID)
+                .setTopic(version < 5 ? "foo" : "")
+                .setPartitions(singletonList(
+                    new OffsetForLeaderEpochRequestData.OffsetForLeaderPartition()
                         .setPartition(0)
                         .setLeaderEpoch(3)
-                        .setCurrentLeaderEpoch(5);
-        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData();
-        data.topics().add(new OffsetForLeaderEpochRequestData.OffsetForLeaderTopic()
-                .setTopic("foo")
-                .setPartitions(singletonList(partitionDataNoCurrentEpoch)));
+                        .setCurrentLeaderEpoch(version >= 2 ? 5 : -1)
+                ))
+        );
 
-        testAllMessageRoundTrips(data);
-        short lowestVersion = ApiKeys.OFFSET_FOR_LEADER_EPOCH.oldestVersion();
-        testAllMessageRoundTripsBetweenVersions(lowestVersion, (short) 2, partitionDataWithCurrentEpoch, partitionDataNoCurrentEpoch);
-        testAllMessageRoundTripsFromVersion((short) 2, partitionDataWithCurrentEpoch);
-
-        // Version 3 adds the optional replica Id field
-        testAllMessageRoundTripsFromVersion((short) 3, new OffsetForLeaderEpochRequestData().setReplicaId(5));
-        testAllMessageRoundTripsBeforeVersion((short) 3,
-                new OffsetForLeaderEpochRequestData().setReplicaId(5),
-                new OffsetForLeaderEpochRequestData());
-        testAllMessageRoundTripsBeforeVersion((short) 3,
-                new OffsetForLeaderEpochRequestData().setReplicaId(5),
-                new OffsetForLeaderEpochRequestData().setReplicaId(-2));
+        testMessageRoundTrip(version, request, request);
     }
 
     @ParameterizedTest
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java
@@ -16,12 +16,18 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderPartition;
+import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopic;
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopicCollection;
 import org.apache.kafka.common.protocol.ApiKeys;
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -51,4 +57,128 @@ public void testForFollower() {
         OffsetsForLeaderEpochRequest parsed = OffsetsForLeaderEpochRequest.parse(request.serialize(), version);
         assertEquals(replicaId, parsed.replicaId());
     }
+
+    @Test
+    public void testVersion4BelowFailUseTopicId() {
+        // Version 4 and below require topic names, not topic IDs
+        OffsetForLeaderTopicCollection topics = new OffsetForLeaderTopicCollection();
+        OffsetForLeaderTopic topic = new OffsetForLeaderTopic()
+                .setTopicId(Uuid.randomUuid())  // Only topicId, no topic name
+                .setPartitions(Collections.singletonList(
+                        new OffsetForLeaderPartition()
+                                .setPartition(0)
+                                .setLeaderEpoch(1)
+                ));
+        topics.add(topic);
+
+        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData()
+                .setReplicaId(1)
+                .setTopics(topics);
+
+        OffsetsForLeaderEpochRequest.Builder builder = new OffsetsForLeaderEpochRequest.Builder(
+                (short) 2, (short) 4, data);
+
+        // Should throw UnsupportedVersionException when building version 4 without topic names
+        assertThrows(UnsupportedVersionException.class, () -> builder.build((short) 4));
+        assertThrows(UnsupportedVersionException.class, () -> builder.build((short) 3));
+        assertThrows(UnsupportedVersionException.class, () -> builder.build((short) 2));
+    }
+
+    @Test
+    public void testVersion4SucceedsWithTopicNames() {
+        // Version 4 should succeed when topic names are provided
+        OffsetForLeaderTopicCollection topics = new OffsetForLeaderTopicCollection();
+        OffsetForLeaderTopic topic = new OffsetForLeaderTopic()
+                .setTopic("test-topic")  // Topic name provided
+                .setPartitions(Collections.singletonList(
+                        new OffsetForLeaderPartition()
+                                .setPartition(0)
+                                .setLeaderEpoch(1)
+                ));
+        topics.add(topic);
+
+        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData()
+                .setReplicaId(1)
+                .setTopics(topics);
+
+        OffsetsForLeaderEpochRequest.Builder builder = new OffsetsForLeaderEpochRequest.Builder(
+                (short) 2, (short) 4, data);
+
+        // Should succeed for version 4 with topic names
+        OffsetsForLeaderEpochRequest request = builder.build((short) 4);
+        assertEquals(1, request.replicaId());
+    }
+
+    @Test
+    public void testVersion5RequiresTopicIds() {
+        // Version 5 and above require topic IDs, not topic names
+        OffsetForLeaderTopicCollection topics = new OffsetForLeaderTopicCollection();
+        OffsetForLeaderTopic topic = new OffsetForLeaderTopic()
+                .setTopic("test-topic")  // Only topic name, no topicId
+                .setPartitions(Collections.singletonList(
+                        new OffsetForLeaderPartition()
+                                .setPartition(0)
+                                .setLeaderEpoch(1)
+                ));
+        topics.add(topic);
+
+        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData()
+                .setReplicaId(-1)
+                .setTopics(topics);
+
+        OffsetsForLeaderEpochRequest.Builder builder = new OffsetsForLeaderEpochRequest.Builder(
+                (short) 3, (short) 5, data);
+
+        // Should throw UnsupportedVersionException when building version 5 without topic IDs
+        assertThrows(UnsupportedVersionException.class, () -> builder.build((short) 5));
+    }
+
+    @Test
+    public void testVersion5SucceedsWithTopicIds() {
+        // Version 5 should succeed when topic IDs are provided
+        OffsetForLeaderTopicCollection topics = new OffsetForLeaderTopicCollection();
+        OffsetForLeaderTopic topic = new OffsetForLeaderTopic()
+                .setTopicId(Uuid.randomUuid())  // Topic ID provided
+                .setPartitions(Collections.singletonList(
+                        new OffsetForLeaderPartition()
+                                .setPartition(0)
+                                .setLeaderEpoch(1)
+                ));
+        topics.add(topic);
+
+        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData()
+                .setReplicaId(-1)
+                .setTopics(topics);
+
+        OffsetsForLeaderEpochRequest.Builder builder = new OffsetsForLeaderEpochRequest.Builder(
+                (short) 3, (short) 5, data);
+
+        // Should succeed for version 5 with topic IDs
+        OffsetsForLeaderEpochRequest request = builder.build((short) 5);
+        assertEquals(-1, request.replicaId());
+    }
+
+    @Test
+    public void testVersion5RejectsZeroTopicId() {
+        // Version 5 should reject ZERO_UUID as topic ID
+        OffsetForLeaderTopicCollection topics = new OffsetForLeaderTopicCollection();
+        OffsetForLeaderTopic topic = new OffsetForLeaderTopic()
+                .setTopicId(Uuid.ZERO_UUID)  // Invalid: ZERO_UUID
+                .setPartitions(Collections.singletonList(
+                        new OffsetForLeaderPartition()
+                                .setPartition(0)
+                                .setLeaderEpoch(1)
+                ));
+        topics.add(topic);
+
+        OffsetForLeaderEpochRequestData data = new OffsetForLeaderEpochRequestData()
+                .setReplicaId(-1)
+                .setTopics(topics);
+
+        OffsetsForLeaderEpochRequest.Builder builder = new OffsetsForLeaderEpochRequest.Builder(
+                (short) 3, (short) 5, data);
+
+        // Should throw UnsupportedVersionException when using ZERO_UUID
+        assertThrows(UnsupportedVersionException.class, () -> builder.build((short) 5));
+    }
 }
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -2117,13 +2117,29 @@ class KafkaApis(val requestChannel: RequestChannel,
     val offsetForLeaderEpoch = request.body[OffsetsForLeaderEpochRequest]
     val topics = offsetForLeaderEpoch.data.topics.asScala.toSeq
 
+    // Separate topics with unknown topic IDs when using version 5+
+    val (knownTopics, unknownTopicIdTopics) = if (OffsetsForLeaderEpochRequest.useTopicIds(request.header.apiVersion)) {
+      topics.partition { offsetForLeaderTopic =>
+        metadataCache.getTopicName(offsetForLeaderTopic.topicId).isPresent
+      }
+    } else {
+      (topics, Seq.empty[OffsetForLeaderTopic])
+    }
+
     // The OffsetsForLeaderEpoch API was initially only used for inter-broker communication and required
     // cluster permission. With KIP-320, the consumer now also uses this API to check for log truncation
     // following a leader change, so we also allow topic describe permission.
     val (authorizedTopics, unauthorizedTopics) =
       if (authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME, logIfDenied = false))
-        (topics, Seq.empty[OffsetForLeaderTopic])
-      else authHelper.partitionSeqByAuthorized(request.context, DESCRIBE, TOPIC, topics)(_.topic)
+        (knownTopics, Seq.empty[OffsetForLeaderTopic])
+      else authHelper.partitionSeqByAuthorized(request.context, DESCRIBE, TOPIC, knownTopics) { offsetForLeaderTopic =>
+        // Resolve topic name from topicId if needed for authorization
+        if (OffsetsForLeaderEpochRequest.useTopicIds(request.header.apiVersion)) {
+          metadataCache.getTopicName(offsetForLeaderTopic.topicId).get()
+        } else {
+          offsetForLeaderTopic.topic
+        }
+      }
 
     val endOffsetsForAuthorizedPartitions = replicaManager.lastOffsetForLeaderEpoch(authorizedTopics)
     val endOffsetsForUnauthorizedPartitions = unauthorizedTopics.map { offsetForLeaderTopic =>
@@ -2133,13 +2149,34 @@ class KafkaApis(val requestChannel: RequestChannel,
           .setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
       }
 
+      // Resolve topic name from topicId if needed
+      val topicName = if (OffsetsForLeaderEpochRequest.useTopicIds(request.header.apiVersion)) {
+        metadataCache.getTopicName(offsetForLeaderTopic.topicId).get()
+      } else {
+        offsetForLeaderTopic.topic
+      }
+
+      new OffsetForLeaderTopicResult()
+        .setTopic(topicName)
+        .setTopicId(offsetForLeaderTopic.topicId)
+        .setPartitions(partitions.toList.asJava)
+    }
+
+    val endOffsetsForUnknownTopicIdTopics = unknownTopicIdTopics.map { offsetForLeaderTopic =>
+      val partitions = offsetForLeaderTopic.partitions.asScala.map { offsetForLeaderPartition =>
+        new EpochEndOffset()
+          .setPartition(offsetForLeaderPartition.partition)
+          .setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
+      }
+
       new OffsetForLeaderTopicResult()
-        .setTopic(offsetForLeaderTopic.topic)
+        .setTopic("")
+        .setTopicId(offsetForLeaderTopic.topicId)
         .setPartitions(partitions.toList.asJava)
     }
 
     val endOffsetsForAllTopics = new OffsetForLeaderTopicResultCollection(
-      (endOffsetsForAuthorizedPartitions ++ endOffsetsForUnauthorizedPartitions).asJava.iterator
+      (endOffsetsForAuthorizedPartitions ++ endOffsetsForUnauthorizedPartitions ++ endOffsetsForUnknownTopicIdTopics).asJava.iterator
     )
 
     requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

Original file line number	Diff line number	Diff line change
`@@ -63,6 +63,10 @@ public Map<Errors, Integer> errorCounts() {`
`63`	`63`	`return errorCounts;`
`64`	`64`	`}`
`65`	`65`
	`66`	`+ public static boolean useTopicIds(short version) {`
	`67`	`+ return version >= 5;`
	`68`	`+ }`
	`69`	`+`
`66`	`70`	`public int throttleTimeMs() {`
`67`	`71`	`return data.throttleTimeMs();`
`68`	`72`	`}`