frankvicky commented on code in PR #19515:
URL: https://github.com/apache/kafka/pull/19515#discussion_r2057673888
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java:
##########
@@ -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);
Review Comment:
Since version 9 is the last version that could use topic names, do we need a
static variable for it?
For example: `LAST_TOPIC_NAME_SUPPORT_VERSION`
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -1834,6 +1834,44 @@ public void testFetchOffsetsWithUnknownGroup() {
assertEquals(expectedResponse, context.fetchOffsets("group", request,
Long.MAX_VALUE));
}
+ @Test
+ public void testFetchOffsetsWithTopicIds() {
+ Uuid fooId = Uuid.randomUuid();
+ Uuid barId = Uuid.randomUuid();
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
+
+
context.groupMetadataManager.getOrMaybeCreatePersistedConsumerGroup("group",
true);
+
+ context.commitOffset("group", "foo", 0, 100L, 1);
+ context.commitOffset("group", "bar", 0, 200L, 1);
Review Comment:
nit: Should we extract `foo`, `bar`, and `group` as variables?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1043,13 +1045,33 @@ class KafkaApis(val requestChannel: RequestChannel,
offsetFetchResponse
} 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.
Review Comment:
IMHO, we could put this comment at the end of the block.
In this way, this comment will be more specific to the condition without the
topic ID.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java:
##########
@@ -1834,6 +1834,44 @@ public void testFetchOffsetsWithUnknownGroup() {
assertEquals(expectedResponse, context.fetchOffsets("group", request,
Long.MAX_VALUE));
}
+ @Test
+ public void testFetchOffsetsWithTopicIds() {
+ Uuid fooId = Uuid.randomUuid();
+ Uuid barId = Uuid.randomUuid();
+ OffsetMetadataManagerTestContext context = new
OffsetMetadataManagerTestContext.Builder().build();
+
+
context.groupMetadataManager.getOrMaybeCreatePersistedConsumerGroup("group",
true);
+
+ context.commitOffset("group", "foo", 0, 100L, 1);
+ context.commitOffset("group", "bar", 0, 200L, 1);
+
+ List<OffsetFetchRequestData.OffsetFetchRequestTopics> request =
Arrays.asList(
Review Comment:
Should use `List.of` instead of `Arrays.List`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
