Skip to content

KAFKA-14691; Add TopicId to OffsetFetch API #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: trunk
Choose a base branch
from
Open

KAFKA-14691; Add TopicId to OffsetFetch API #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
775e2eb
update request/response
dajac Apr 18, 2025
c542dc3
update kafkapis
dajac Apr 18, 2025
3fdd4e6
wip
dajac Apr 18, 2025
5ebe55e
tests
dajac Apr 18, 2025
8c11aee
unit test
dajac Apr 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.common.requests;

import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.OffsetFetchRequestData;
import org.apache.kafka.common.message.OffsetFetchRequestData.OffsetFetchRequestGroup;
Expand Down Expand Up @@ -71,7 +72,8 @@ public Builder(String groupId,
boolean requireStable,
List<TopicPartition> partitions,
boolean throwOnFetchStableOffsetsUnsupported) {
super(ApiKeys.OFFSET_FETCH);
// It can only be used with topic names.
super(ApiKeys.OFFSET_FETCH, ApiKeys.OFFSET_FETCH.oldestVersion(), (short) 9);

OffsetFetchRequestData.OffsetFetchRequestGroup group =
new OffsetFetchRequestData.OffsetFetchRequestGroup()
Expand Down Expand Up @@ -103,7 +105,8 @@ public Builder(String groupId,
public Builder(Map<String, List<TopicPartition>> groupIdToTopicPartitionMap,
boolean requireStable,
boolean throwOnFetchStableOffsetsUnsupported) {
super(ApiKeys.OFFSET_FETCH);
// It can only be used with topic names.
super(ApiKeys.OFFSET_FETCH, ApiKeys.OFFSET_FETCH.oldestVersion(), (short) 9);

List<OffsetFetchRequestGroup> groups = new ArrayList<>();
for (Entry<String, List<TopicPartition>> entry : groupIdToTopicPartitionMap.entrySet()) {
Expand Down Expand Up @@ -134,6 +137,12 @@ public Builder(Map<String, List<TopicPartition>> groupIdToTopicPartitionMap,
this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported;
}

public Builder(OffsetFetchRequestData data, boolean throwOnFetchStableOffsetsUnsupported) {
super(ApiKeys.OFFSET_FETCH);
this.data = data;
this.throwOnFetchStableOffsetsUnsupported = throwOnFetchStableOffsetsUnsupported;
}

@Override
public OffsetFetchRequest build(short version) {
if (data.groups().size() > 1 && version < 8) {
Expand Down Expand Up @@ -350,4 +359,8 @@ public boolean isAllPartitionsForGroup(String groupId) {
public OffsetFetchRequestData data() {
return data;
}

public static boolean useTopicIds(short version) {
return version >= 10;
}
}
9 changes: 7 additions & 2 deletions clients/src/main/resources/common/message/OffsetFetchRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,11 @@
//
// Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). It adds
// the MemberId and MemberEpoch fields. Those are filled in and validated when the new consumer protocol is used.
"validVersions": "1-9",
//
// Version 10 adds support for topic ids (KIP-848).
"validVersions": "1-10",
"flexibleVersions": "6+",
"latestVersionUnstable": true,
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0-7", "entityType": "groupId",
"about": "The group to fetch offsets for." },
Expand All @@ -60,8 +63,10 @@
"about": "The member epoch if using the new consumer protocol (KIP-848)." },
{ "name": "Topics", "type": "[]OffsetFetchRequestTopics", "versions": "8+", "nullableVersions": "8+",
"about": "Each topic we would like to fetch offsets for, or null to fetch offsets for all topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "8+", "entityType": "topicName",
{ "name": "Name", "type": "string", "versions": "8-9", "entityType": "topicName", "ignorable": true,
"about": "The topic name."},
{ "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
"about": "The topic ID." },
{ "name": "PartitionIndexes", "type": "[]int32", "versions": "8+",
"about": "The partition indexes we would like to fetch offsets for." }
]}
Expand Down
9 changes: 7 additions & 2 deletions clients/src/main/resources/common/message/OffsetFetchResponse.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@
// Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The response is
// the same as version 8 but can return STALE_MEMBER_EPOCH and UNKNOWN_MEMBER_ID errors when the new consumer group
// protocol is used.
"validVersions": "1-9",
//
// Version 10 adds support for topic ids (KIP-848).
"validVersions": "1-10",
"flexibleVersions": "6+",
// Supported errors:
// - GROUP_AUTHORIZATION_FAILED (version 0+)
Expand All @@ -49,6 +51,7 @@
// - UNSTABLE_OFFSET_COMMIT (version 7+)
// - UNKNOWN_MEMBER_ID (version 9+)
// - STALE_MEMBER_EPOCH (version 9+)
// - UNKNOWN_TOPIC_ID (version 10+)
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "3+", "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." },
Expand Down Expand Up @@ -78,8 +81,10 @@
"about": "The group ID." },
{ "name": "Topics", "type": "[]OffsetFetchResponseTopics", "versions": "8+",
"about": "The responses per topic.", "fields": [
{ "name": "Name", "type": "string", "versions": "8+", "entityType": "topicName",
{ "name": "Name", "type": "string", "versions": "8-9", "entityType": "topicName", "ignorable": true,
"about": "The topic name." },
{ "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
"about": "The topic ID." },
{ "name": "Partitions", "type": "[]OffsetFetchResponsePartitions", "versions": "8+",
"about": "The responses per partition.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "8+",
Expand Down
80 changes: 66 additions & 14 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1027,6 +1027,8 @@ class KafkaApis(val requestChannel: RequestChannel,
offsetFetchRequest: OffsetFetchRequestData.OffsetFetchRequestGroup,
requireStable: Boolean
): CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup] = {
val useTopicIds = OffsetFetchRequest.useTopicIds(requestContext.apiVersion)

groupCoordinator.fetchAllOffsets(
requestContext,
offsetFetchRequest,
Expand All @@ -1040,13 +1042,33 @@ class KafkaApis(val requestChannel: RequestChannel,
offsetFetchResponse
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Authorization Logic with Topic IDs: This section modifies authorization logic to handle topic IDs. Please confirm that authorization checks work properly when using topic IDs instead of topic names, as this is security-critical functionality.

} else {
// Clients are not allowed to see offsets for topics that are not authorized for Describe.
val (authorizedOffsets, _) = authHelper.partitionSeqByAuthorized(
val authorizedNames = authHelper.filterByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
offsetFetchResponse.topics.asScala
)(_.name)
offsetFetchResponse.setTopics(authorizedOffsets.asJava)

val topics = new mutable.ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseTopics]
offsetFetchResponse.topics.forEach { topic =>
if (authorizedNames.contains(topic.name)) {
if (useTopicIds) {
// If the topic is not provided by the group coordinator, we set it
// using the metadata cache.
if (topic.topicId == Uuid.ZERO_UUID) {
topic.setTopicId(metadataCache.getTopicId(topic.name))
}
// If we don't have the topic id at all, we skip the topic because
// we can not serialize it without it.
if (topic.topicId != Uuid.ZERO_UUID) {
topics += topic
}
} else {
topics += topic
}
}
}
offsetFetchResponse.setTopics(topics.asJava)
}
}
}
Expand All @@ -1056,14 +1078,53 @@ class KafkaApis(val requestChannel: RequestChannel,
offsetFetchRequest: OffsetFetchRequestData.OffsetFetchRequestGroup,
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Topic ID Support Version Handling: Please verify that the implementation correctly handles both topic IDs and topic names across different API versions, especially around version 10 which introduces topic ID support. The conditional logic paths based on version checks need careful verification to ensure backward compatibility is maintained.

requireStable: Boolean
): CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup] = {
val useTopicIds = OffsetFetchRequest.useTopicIds(requestContext.apiVersion)

if (useTopicIds) {
offsetFetchRequest.topics.forEach { topic =>
if (topic.topicId != Uuid.ZERO_UUID) {
metadataCache.getTopicName(topic.topicId).ifPresent(name => topic.setName(name))
}
}
}

// Clients are not allowed to see offsets for topics that are not authorized for Describe.
val (authorizedTopics, unauthorizedTopics) = authHelper.partitionSeqByAuthorized(
val authorizedTopicNames = authHelper.filterByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
offsetFetchRequest.topics.asScala
)(_.name)

val authorizedTopics = new mutable.ArrayBuffer[OffsetFetchRequestData.OffsetFetchRequestTopics]
val errorTopics = new mutable.ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseTopics]

def buildErrorResponse(
topic: OffsetFetchRequestData.OffsetFetchRequestTopics,
error: Errors
): OffsetFetchResponseData.OffsetFetchResponseTopics = {
val topicResponse = new OffsetFetchResponseData.OffsetFetchResponseTopics()
.setTopicId(topic.topicId)
.setName(topic.name)
topic.partitionIndexes.forEach { partitionIndex =>
topicResponse.partitions.add(new OffsetFetchResponseData.OffsetFetchResponsePartitions()
.setPartitionIndex(partitionIndex)
.setCommittedOffset(-1)
.setErrorCode(error.code))
}
topicResponse
}

offsetFetchRequest.topics.forEach { topic =>
if (useTopicIds && topic.name.isEmpty) {
errorTopics += buildErrorResponse(topic, Errors.UNKNOWN_TOPIC_ID)
} else if (!authorizedTopicNames.contains(topic.name)) {
errorTopics += buildErrorResponse(topic, Errors.TOPIC_AUTHORIZATION_FAILED)
} else {
authorizedTopics += topic
}
}

groupCoordinator.fetchOffsets(
requestContext,
new OffsetFetchRequestData.OffsetFetchRequestGroup()
Expand All @@ -1081,19 +1142,10 @@ class KafkaApis(val requestChannel: RequestChannel,
offsetFetchResponse
} else {
val topics = new util.ArrayList[OffsetFetchResponseData.OffsetFetchResponseTopics](
offsetFetchResponse.topics.size + unauthorizedTopics.size
offsetFetchResponse.topics.size + errorTopics.size
)
topics.addAll(offsetFetchResponse.topics)
unauthorizedTopics.foreach { topic =>
val topicResponse = new OffsetFetchResponseData.OffsetFetchResponseTopics().setName(topic.name)
topic.partitionIndexes.forEach { partitionIndex =>
topicResponse.partitions.add(new OffsetFetchResponseData.OffsetFetchResponsePartitions()
.setPartitionIndex(partitionIndex)
.setCommittedOffset(-1)
.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code))
}
topics.add(topicResponse)
}
topics.addAll(errorTopics.asJava)
offsetFetchResponse.setTopics(topics)
}
}
Expand Down
Loading