dajac commented on code in PR #19515:
URL: https://github.com/apache/kafka/pull/19515#discussion_r2085262516
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -8349,23 +8377,168 @@ class KafkaApisTest extends Logging {
.setGroupId("group-4")
.setErrorCode(Errors.INVALID_GROUP_ID.code)
- val expectedGroups = List(group1Response, group2Response,
group3Response, group4Response)
+ val expectedGroups = List(expectedGroup1Response, group2Response,
group3Response, group4Response)
group1Future.complete(group1Response)
group2Future.complete(group2Response)
group3Future.completeExceptionally(Errors.INVALID_GROUP_ID.exception)
group4Future.complete(group4Response)
val response =
verifyNoThrottling[OffsetFetchResponse](requestChannelRequest)
- assertEquals(expectedGroups.toSet, response.data.groups().asScala.toSet)
+ assertEquals(expectedGroups.toSet, response.data.groups.asScala.toSet)
}
}
@ParameterizedTest
- @ApiKeyVersionsSource(apiKey = ApiKeys.OFFSET_FETCH)
+ // We only test with topic ids.
+ @ApiKeyVersionsSource(apiKey = ApiKeys.OFFSET_FETCH, fromVersion = 10)
+ def testHandleOffsetFetchWithUnknownTopicIds(version: Short): Unit = {
+ val foo = "foo"
+ val bar = "bar"
+ val fooId = Uuid.randomUuid()
+ val barId = Uuid.randomUuid()
+ addTopicToMetadataCache(foo, topicId = fooId, numPartitions = 2)
+
+ def makeRequest(version: Short): RequestChannel.Request = {
+ buildRequest(
+ OffsetFetchRequest.Builder.forTopicIdsOrNames(
+ new OffsetFetchRequestData()
+ .setGroups(List(
+ new OffsetFetchRequestData.OffsetFetchRequestGroup()
+ .setGroupId("group-1")
+ .setTopics(List(
+ new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ .setName(foo)
+ .setTopicId(fooId)
+ .setPartitionIndexes(List[Integer](0).asJava),
+ // bar does not exist so it must return UNKNOWN_TOPIC_ID.
+ new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ .setName(bar)
+ .setTopicId(barId)
+ .setPartitionIndexes(List[Integer](0).asJava)
+ ).asJava),
+ new OffsetFetchRequestData.OffsetFetchRequestGroup()
+ .setGroupId("group-2")
+ .setTopics(null)
+ ).asJava),
+ false,
+ true
+ ).build(version)
+ )
+ }
+
+ val requestChannelRequest = makeRequest(version)
+
+ val group1Future = new
CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup]()
+ when(groupCoordinator.fetchOffsets(
+ requestChannelRequest.context,
+ new OffsetFetchRequestData.OffsetFetchRequestGroup()
+ .setGroupId("group-1")
+ .setTopics(List(
+ new OffsetFetchRequestData.OffsetFetchRequestTopics()
+ .setTopicId(fooId)
+ .setName("foo")
+ .setPartitionIndexes(List[Integer](0).asJava)).asJava),
+ false
+ )).thenReturn(group1Future)
+
+ val group2Future = new
CompletableFuture[OffsetFetchResponseData.OffsetFetchResponseGroup]()
+ when(groupCoordinator.fetchAllOffsets(
+ requestChannelRequest.context,
+ new OffsetFetchRequestData.OffsetFetchRequestGroup()
+ .setGroupId("group-2")
+ .setTopics(null),
+ false
+ )).thenReturn(group2Future)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val group1Response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
+ .setGroupId("group-1")
+ .setTopics(List(
+ new OffsetFetchResponseData.OffsetFetchResponseTopics()
+ .setTopicId(fooId)
+ .setName(foo)
+ .setPartitions(List(
+ new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+ .setPartitionIndex(0)
+ .setCommittedOffset(100)
+ .setCommittedLeaderEpoch(1)
+ ).asJava)
+ ).asJava)
+
+ val group2Response = new OffsetFetchResponseData.OffsetFetchResponseGroup()
+ .setGroupId("group-2")
+ .setTopics(List(
+ new OffsetFetchResponseData.OffsetFetchResponseTopics()
+ .setName(foo)
+ .setPartitions(List(
+ new OffsetFetchResponseData.OffsetFetchResponsePartitions()
+ .setPartitionIndex(0)
+ .setCommittedOffset(100)
+ .setCommittedLeaderEpoch(1)
+ ).asJava),
+ // 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)
Review Comment:
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).
--
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
