jeffkbkim commented on code in PR #14124:
URL: https://github.com/apache/kafka/pull/14124#discussion_r1300435377
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3661,6 +3662,37 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+ def handleConsumerGroupDescribe(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ // TODO: This method is Work-In-Progress
+ // It will be finished in the second PR of KAFKA-14509
+ val consumerGroupDescribeRequest =
request.body[ConsumerGroupDescribeRequest]
+
+ if (!config.isNewGroupCoordinatorEnabled) {
+ // The API is not supported by the "old" group coordinator (the
default). If the
+ // new one is not enabled, we fail directly here.
+ requestHelper.sendMaybeThrottle(request,
consumerGroupDescribeRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else {
+
+// val response = new ConsumerGroupDescribeResponseData()
Review Comment:
nit: do we need this?
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3661,6 +3662,37 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+ def handleConsumerGroupDescribe(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ // TODO: This method is Work-In-Progress
+ // It will be finished in the second PR of KAFKA-14509
+ val consumerGroupDescribeRequest =
request.body[ConsumerGroupDescribeRequest]
+
+ if (!config.isNewGroupCoordinatorEnabled) {
+ // The API is not supported by the "old" group coordinator (the
default). If the
+ // new one is not enabled, we fail directly here.
+ requestHelper.sendMaybeThrottle(request,
consumerGroupDescribeRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else {
+
+// val response = new ConsumerGroupDescribeResponseData()
+ consumerGroupDescribeRequest.data().groupIds().forEach { groupId =>
+ val describedGroup = new
ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+
+ // TODO: Check if group id exists
+
+ if (!authHelper.authorize(request.context, READ, GROUP, groupId)) {
+ describedGroup.setErrorMessage(new
ApiError(Errors.GROUP_AUTHORIZATION_FAILED).message())
+ describedGroup.setErrorCode(Errors.GROUP_AUTHORIZATION_FAILED.code())
+ }
+ }
+
+// This must be changed to handle all groups together instead of foreach
Review Comment:
the comment indentation is off
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3661,6 +3662,37 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+ def handleConsumerGroupDescribe(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ // TODO: This method is Work-In-Progress
+ // It will be finished in the second PR of KAFKA-14509
+ val consumerGroupDescribeRequest =
request.body[ConsumerGroupDescribeRequest]
+
+ if (!config.isNewGroupCoordinatorEnabled) {
+ // The API is not supported by the "old" group coordinator (the
default). If the
+ // new one is not enabled, we fail directly here.
+ requestHelper.sendMaybeThrottle(request,
consumerGroupDescribeRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else {
+
+// val response = new ConsumerGroupDescribeResponseData()
+ consumerGroupDescribeRequest.data().groupIds().forEach { groupId =>
+ val describedGroup = new
ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+
+ // TODO: Check if group id exists
+
+ if (!authHelper.authorize(request.context, READ, GROUP, groupId)) {
+ describedGroup.setErrorMessage(new
ApiError(Errors.GROUP_AUTHORIZATION_FAILED).message())
+ describedGroup.setErrorCode(Errors.GROUP_AUTHORIZATION_FAILED.code())
+ }
+ }
+
+// This must be changed to handle all groups together instead of foreach
+// groupCoordinator.describeGroups(request.context, )
Review Comment:
can we also remove this?
--
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
