KAFKA-14691; Add TopicId to OffsetFetch API #19515

dajac · 2025-04-19T13:51:53Z

This patch extends the OffsetFetch API to support topic ids. From
version 10 of the API, topic ids must be used.

The patch only contains the server side changes and it keeps the version
10 as unstable for now. We will mark the version as stable when the
client side changes are merged in.

Reviewers: TengYao Chi frankvicky@apache.org, Lianet Magrans
lmagrans@confluent.io

frankvicky

@dajac: Thanks for the patch!
I have a few minor comments for consideration.

core/src/main/scala/kafka/server/KafkaApis.scala

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

...-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java

clients/src/test/java/org/apache/kafka/common/message/MessageTest.java

dajac · 2025-04-25T16:02:06Z

This PR is getting out of control because OffsetFetchRequest and OffsetFetchResponse are way too complicated. I will simplify them in separate PRs and come back to this one afterwards.

dajac · 2025-04-26T07:35:52Z

Opened #19572.

dajac · 2025-04-26T19:49:59Z

Opened #19576.

While working on #19515, I came to the conclusion that the OffsetFetchRequest is quite messy and overall too complicated. This patch rationalize the constructors. OffsetFetchRequest has a single constructor accepting the OffsetFetchRequestData. This will also simplify adding the topic ids. All the changes are mechanical, replacing data structures by others. Reviewers: PoAn Yang <payang@apache.org>, TengYao Chi <frankvicky@apache.org>, Lianet Magran <lmagrans@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

dajac · 2025-05-06T06:54:54Z

Opened #19642.

While working on #19515, I came to the conclusion that the OffsetFetchResponse is quite messy and overall too complicated. This patch rationalize the constructors. OffsetFetchResponse has a single constructor accepting the OffsetFetchResponseData. A builder is introduced to handle the down conversion. This will also simplify adding the topic ids. All the changes are mechanical, replacing data structures by others. Reviewers: Lianet Magrans <lmagrans@confluent.io>

clients/src/test/java/org/apache/kafka/common/requests/OffsetFetchRequestTest.java

lianetm

Thanks for the changes! Just some minor comments.

lianetm · 2025-05-12T15:39:27Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

            this.data = data;
            this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported;
        }

        @Override
+        @SuppressWarnings("checkstyle:cyclomaticComplexity")


would it help if we move the logic to validate topic id and names usage to a different func? (this is wild indeed, maybe we could consider also encapsulating the other validation sections consistently: batching, require stable)

lianetm · 2025-05-12T16:08:08Z

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

+                .setTopicId(topicId)
+                .setPartitionIndexes(List[Integer](0, 1, 2).asJava)
+            ).asJava),
+          requireStable = true,


is this change to true intentional?

nope, let me revert it.

lianetm · 2025-05-12T16:16:25Z

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

@@ -8269,6 +8276,7 @@ class KafkaApisTest extends Logging {
          .setGroupId("group-1")
          .setTopics(List(
            new OffsetFetchRequestData.OffsetFetchRequestTopics()
+              .setTopicId(if (version >= 10) fooId else Uuid.ZERO_UUID)
              .setName("foo")


shouldn't we set the name in the request only if v<10?

nope because this is the response from the group coordinator service. the topic name is always set but the topic id is only set if version >= 10.

lianetm · 2025-05-12T16:18:18Z

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

@@ -8320,11 +8347,12 @@ class KafkaApisTest extends Logging {
            ).asJava)
        ).asJava)

+


nit: extra line

lianetm · 2025-05-12T16:57:02Z

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

+        // bar does not exist so it must be filtered out.
+        new OffsetFetchResponseData.OffsetFetchResponseTopics()
+          .setName(bar)
+          .setPartitions(List(
+            new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+              .setPartitionIndex(0)
+              .setCommittedOffset(100)
+              .setCommittedLeaderEpoch(1)
+          ).asJava)


just for my understanding, this would be the case where bar does exist when the coordinator fetched the offsets, but by the time we build the response in KafkaApis it doesn't exist anymore (so filtered out), correct?

almost. we requested all topics for group-2. in this case, we resolve the topic id when the resolve is built. if the topic id does not exist, we drop the topic (it does not exist).

lianetm

Thanks for the updates! LGTM

github-actions bot added core Kafka Broker clients labels Apr 19, 2025

dajac added 6 commits April 22, 2025 20:59

update request/response

5e3b451

update kafkapis

c57bda7

wip

ad6e110

tests

0586a56

unit test

7f4dcb4

refactor

f76603a

dajac force-pushed the KAFKA-14691 branch from 8c11aee to f76603a Compare April 22, 2025 19:03

dajac added 4 commits April 22, 2025 21:18

fix imports

35fb001

check topic id and name in builder

9bf1cd9

Merge remote-tracking branch 'upstream/trunk' into KAFKA-14691

fbc88d2

update integration tests

0177895

dajac marked this pull request as ready for review April 23, 2025 19:57

dajac added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Apr 23, 2025

fix comment

a38e2eb

dajac requested a review from lianetm April 23, 2025 20:02

fix checkstyle

4e19cce

frankvicky reviewed Apr 24, 2025

View reviewed changes

dajac added 3 commits April 24, 2025 21:17

address minor comments

ab59a1b

fix tests

49dd71a

fix

22e00ba

dajac commented Apr 24, 2025

View reviewed changes

clients/src/test/java/org/apache/kafka/common/message/MessageTest.java Outdated Show resolved Hide resolved

dajac added 3 commits April 25, 2025 17:31

cleanup MessageTest

6187ce6

remove unused var

c070a95

refactor get error response

ef63234

dajac mentioned this pull request Apr 26, 2025

MINOR: Simplify OffsetFetchRequest #19572

Merged

dajac mentioned this pull request May 6, 2025

MINOR: Simplify OffsetFetchResponse #19642

Merged

dajac added 2 commits May 9, 2025 11:04

Merge remote-tracking branch 'upstream/trunk' into KAFKA-14691

9757444

fixes

1f3b3d6

github-actions bot added the consumer label May 9, 2025

update tests

6a89760

dajac requested a review from frankvicky May 10, 2025 03:58

frankvicky reviewed May 12, 2025

View reviewed changes

clients/src/test/java/org/apache/kafka/common/requests/OffsetFetchRequestTest.java Outdated Show resolved Hide resolved

remove duplicate setTopicId

6663af0

lianetm reviewed May 12, 2025

View reviewed changes

dajac added 2 commits May 12, 2025 20:54

minor fixes

b3718eb

refactor

86173b6

dajac requested a review from lianetm May 12, 2025 19:13

lianetm approved these changes May 13, 2025

View reviewed changes

dajac merged commit d154a31 into apache:trunk May 13, 2025
25 checks passed

dajac deleted the KAFKA-14691 branch May 13, 2025 13:10

		@@ -8320,11 +8347,12 @@ class KafkaApisTest extends Logging {
		).asJava)
		).asJava)

KAFKA-14691; Add TopicId to OffsetFetch API #19515

KAFKA-14691; Add TopicId to OffsetFetch API #19515

Uh oh!

Conversation

dajac commented Apr 19, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

dajac commented Apr 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dajac commented Apr 26, 2025

Uh oh!

dajac commented Apr 26, 2025

Uh oh!

dajac commented May 6, 2025

Uh oh!

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dajac commented Apr 19, 2025 •

edited by github-actions bot

Loading

dajac commented Apr 25, 2025 •

edited

Loading