apoorvmittal10 commented on code in PR #18834:
URL: https://github.com/apache/kafka/pull/18834#discussion_r1954387770
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3190,26 +3190,50 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleDescribeShareGroupOffsetsRequest(request: RequestChannel.Request):
Unit = {
val describeShareGroupOffsetsRequest =
request.body[DescribeShareGroupOffsetsRequest]
+ val groups = describeShareGroupOffsetsRequest.groups()
+
+ val futures = new
mutable.ArrayBuffer[CompletableFuture[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup]](groups.size)
+ groups.forEach { groupDescribeOffsets =>
+ if (!isShareGroupProtocolEnabled) {
+ futures += CompletableFuture.completedFuture(new
DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
+ .setGroupId(groupDescribeOffsets.groupId)
+ .setErrorCode(Errors.UNSUPPORTED_VERSION.code))
+ } else if (!authHelper.authorize(request.context, READ, GROUP,
groupDescribeOffsets.groupId)) {
+ futures += CompletableFuture.completedFuture(new
DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
+ .setGroupId(groupDescribeOffsets.groupId)
+ .setErrorCode(Errors.GROUP_AUTHORIZATION_FAILED.code))
+ } else if (groupDescribeOffsets.topics.isEmpty) {
+ futures += CompletableFuture.completedFuture(new
DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup()
+ .setGroupId(groupDescribeOffsets.groupId))
+ } else {
+ futures += describeShareGroupOffsetsForGroup(
+ request.context,
+ groupDescribeOffsets
+ )
+ }
+ }
- if (!isShareGroupProtocolEnabled) {
- requestHelper.sendMaybeThrottle(request,
describeShareGroupOffsetsRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
- CompletableFuture.completedFuture[Unit](())
- } else if (!authHelper.authorize(request.context, READ, GROUP,
describeShareGroupOffsetsRequest.data.groupId)) {
- requestHelper.sendMaybeThrottle(request,
describeShareGroupOffsetsRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
- CompletableFuture.completedFuture[Unit](())
- } else {
- groupCoordinator.describeShareGroupOffsets(
- request.context,
- describeShareGroupOffsetsRequest.data,
- ).handle[Unit] { (response, exception) =>
- if (exception != null) {
- requestHelper.sendMaybeThrottle(request,
describeShareGroupOffsetsRequest.getErrorResponse(exception))
- } else {
- requestHelper.sendMaybeThrottle(
- request,
- new DescribeShareGroupOffsetsResponse(response)
- )
- }
+ CompletableFuture.allOf(futures.toArray: _*).handle[Unit] { (_, _) =>
+ val groupResponses = new
ArrayBuffer[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup](futures.size)
+ val responseData = new
DescribeShareGroupOffsetsResponseData().setGroups(groupResponses.asJava)
+ futures.foreach(future => groupResponses += future.get())
Review Comment:
As per the API of `allOf` suggests to use join.
```suggestion
futures.foreach(future => groupResponses += future.join)
```
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -10558,7 +10582,81 @@ class KafkaApisTest extends Logging {
))
))
- future.complete(describeShareGroupOffsetsResponse)
+ val describeShareGroupOffsetsResponseGroup2 = new
DescribeShareGroupOffsetsResponseGroup()
+ .setGroupId("group2")
+ .setTopics(util.List.of(
+ new DescribeShareGroupOffsetsResponseTopic()
+ .setTopicName(topicName3)
+ .setTopicId(topicId3)
+ .setPartitions(util.List.of(
+ new DescribeShareGroupOffsetsResponsePartition()
+ .setPartitionIndex(0)
+ .setStartOffset(0)
+ .setLeaderEpoch(1)
+ .setErrorMessage(null)
+ .setErrorCode(0)
+ ))
+ ))
+
+ val describeShareGroupOffsetsResponse = new
DescribeShareGroupOffsetsResponseData()
+ .setGroups(util.List.of(describeShareGroupOffsetsResponseGroup1,
describeShareGroupOffsetsResponseGroup2))
+
+ futureGroup1.complete(describeShareGroupOffsetsResponseGroup1)
+ futureGroup2.complete(describeShareGroupOffsetsResponseGroup2)
+ val response =
verifyNoThrottling[DescribeShareGroupOffsetsResponse](requestChannelRequest)
+ assertEquals(describeShareGroupOffsetsResponse, response.data)
+ }
+
+ @Test
+ def testDescribeShareGroupOffsetsRequestEmptyGroupsSuccess(): Unit = {
+ metadataCache = MetadataCache.kRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_0)
+
+ val describeShareGroupOffsetsRequest = new
DescribeShareGroupOffsetsRequestData()
Review Comment:
```suggestion
val describeShareGroupOffsetsRequest = new
DescribeShareGroupOffsetsRequestData
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
