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)
Comment on lines +1045 to 1050
Copy link

Choose a reason for hiding this comment

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

Null Check Missing

No null check on offsetFetchResponse.topics before conversion to Scala collection. Could cause NullPointerException if topics is null, enabling denial of service attacks.

Suggested change
val authorizedNames = authHelper.filterByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
offsetFetchResponse.topics.asScala
)(_.name)
val authorizedNames = if (offsetFetchResponse.topics == null) {
Set.empty[String]
} else {
authHelper.filterByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
offsetFetchResponse.topics.asScala
)(_.name)
}
Standards
  • CWE-476
  • OWASP-A06

Comment on lines +1045 to 1050
Copy link

Choose a reason for hiding this comment

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

Unnecessary String Comparisons

Filtering by topic names causes unnecessary string comparisons when topic IDs are available. With many topics, this creates O(n) lookups where O(1) is possible using topic IDs directly.

Suggested change
val authorizedNames = authHelper.filterByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
offsetFetchResponse.topics.asScala
)(_.name)
val authorizedIdentifiers = if (useTopicIds) {
authHelper.filterByAuthorized(requestContext, DESCRIBE, TOPIC, offsetFetchResponse.topics.asScala)(topic => metadataCache.getTopicName(topic.topicId).orElse(topic.name))
} else {
authHelper.filterByAuthorized(requestContext, DESCRIBE, TOPIC, offsetFetchResponse.topics.asScala)(_.name)
}
Standards
  • ISO-IEC-25010-Performance-Time-Behaviour
  • Algorithm-Opt-Hash-Map
  • Netflix-Hot-Path-Optimization

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))
}
}
Comment on lines +1084 to +1088
Copy link

Choose a reason for hiding this comment

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

Null Check Missing

Missing null check on offsetFetchRequest.topics could cause NullPointerException. Attackers could send malformed requests with null topics to trigger server errors.

Suggested change
offsetFetchRequest.topics.forEach { topic =>
if (topic.topicId != Uuid.ZERO_UUID) {
metadataCache.getTopicName(topic.topicId).ifPresent(name => topic.setName(name))
}
}
if (useTopicIds) {
if (offsetFetchRequest.topics != null) {
offsetFetchRequest.topics.forEach { topic =>
if (topic.topicId != Uuid.ZERO_UUID) {
metadataCache.getTopicName(topic.topicId).ifPresent(name => topic.setName(name))
}
}
}
}
Standards
  • CWE-476
  • OWASP-A06

Comment on lines +1082 to +1088
Copy link

Choose a reason for hiding this comment

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

Topic ID Inconsistency

Topic name resolution only happens when topicId is non-zero, but later code at line 1119 checks for empty name regardless of ID value. This creates inconsistent validation where topics with ZERO_UUID but empty names trigger UNKNOWN_TOPIC_ID error incorrectly.

Standards
  • Logic-Verification-Consistency
  • Business-Rule-Validation
  • Algorithm-Correctness-Conditional

}
Comment on lines +1083 to +1089
Copy link

Choose a reason for hiding this comment

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

Missing TopicId Validation

Missing validation when topic ID is present but name lookup fails. This creates inconsistency where topicId exists but name remains empty, causing potential reliability issues.

Suggested change
if (useTopicIds) {
offsetFetchRequest.topics.forEach { topic =>
if (topic.topicId != Uuid.ZERO_UUID) {
metadataCache.getTopicName(topic.topicId).ifPresent(name => topic.setName(name))
}
}
}
if (useTopicIds) {
offsetFetchRequest.topics.forEach { topic =>
if (topic.topicId != Uuid.ZERO_UUID) {
val topicNameOpt = metadataCache.getTopicName(topic.topicId)
if (topicNameOpt.isPresent) {
topic.setName(topicNameOpt.get)
} else {
// If we can't resolve the topic ID, mark it with empty name
// It will be handled later with UNKNOWN_TOPIC_ID error
topic.setName("")
}
}
}
}
Standards
  • ISO-IEC-25010-Reliability-Fault-Tolerance
  • ISO-IEC-25010-Functional-Correctness-Appropriateness
  • DbC-Precondition-Validation
  • SRE-Error-Handling

Comment on lines +1082 to +1089
Copy link

Choose a reason for hiding this comment

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

Redundant Topic Name

Setting topic names when using topic IDs creates unnecessary overhead. When using topic IDs (version >= 10), the name isn't required for lookups. This redundant operation adds CPU and memory overhead by populating fields that won't be used in the authorization path.

Standards
  • ISO-IEC-25010-Performance-Resource-Utilization
  • Algorithm-Opt-Redundant-Operation


// 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
}
Comment on lines +1102 to +1116
Copy link

Choose a reason for hiding this comment

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

Incomplete Error Handling

Error response builder doesn't set committed leader epoch. This inconsistency could lead to client-side errors when processing responses with errors, as other code paths set this field.

Suggested change
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
}
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)
.setCommittedLeaderEpoch(-1)
.setErrorCode(error.code))
}
topicResponse
}
Standards
  • ISO-IEC-25010-Reliability-Maturity
  • ISO-IEC-25010-Functional-Correctness-Completeness
  • DbC-Postcondition-Consistency
  • SRE-Error-Handling


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)
Comment on lines +1118 to +1122
Copy link

Choose a reason for hiding this comment

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

Incomplete Error Handling

Logic fails to handle case where topic.name is non-empty but invalid/non-existent. Authorization check will fail but with incorrect error code (TOPIC_AUTHORIZATION_FAILED instead of UNKNOWN_TOPIC_ID) when topic ID is provided but resolves to non-existent name.

Standards
  • Business-Rule-Validation
  • Logic-Verification-Completeness
  • Algorithm-Correctness-ErrorHandling

} else {
authorizedTopics += topic
}
}
Comment on lines +1099 to +1126
Copy link

Choose a reason for hiding this comment

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

Inefficient Collection Conversion

Multiple collection conversions and intermediate buffers create unnecessary overhead. The code creates temporary buffers and performs multiple iterations over the topic collection. Consider using partition/filter operations to categorize topics in a single pass, reducing memory allocations and CPU cycles.

Standards
  • ISO-IEC-25010-Performance-Resource-Utilization
  • Algorithm-Opt-Collection-Processing


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
Loading