bachmanity1 commented on code in PR #14306:
URL: https://github.com/apache/kafka/pull/14306#discussion_r1723861428
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3510,43 +3510,34 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleDescribeCluster(request: RequestChannel.Request): Unit = {
- val describeClusterRequest = request.body[DescribeClusterRequest]
-
- var clusterAuthorizedOperations = Int.MinValue // Default value in the
schema
- // get cluster authorized operations
- if (describeClusterRequest.data.includeClusterAuthorizedOperations) {
- if (authHelper.authorize(request.context, DESCRIBE, CLUSTER,
CLUSTER_NAME))
- clusterAuthorizedOperations = authHelper.authorizedOperations(request,
Resource.CLUSTER)
- else
- clusterAuthorizedOperations = 0
- }
-
- val brokers =
metadataCache.getAliveBrokerNodes(request.context.listenerName)
- val controllerId = {
- metadataCache.getControllerId.flatMap {
- case ZkCachedControllerId(id) => Some(id)
- case KRaftCachedControllerId(_) => metadataCache.getRandomAliveBrokerId
- }
- }
-
- requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => {
- val data = new DescribeClusterResponseData()
- .setThrottleTimeMs(requestThrottleMs)
- .setClusterId(clusterId)
-
.setControllerId(controllerId.getOrElse(MetadataResponse.NO_CONTROLLER_ID))
- .setClusterAuthorizedOperations(clusterAuthorizedOperations)
-
-
- brokers.foreach { broker =>
- data.brokers.add(new
DescribeClusterResponseData.DescribeClusterBroker()
- .setBrokerId(broker.id)
- .setHost(broker.host)
- .setPort(broker.port)
- .setRack(broker.rack))
+ val response = authHelper.computeDescribeClusterResponse(
+ request,
+ EndpointType.BROKER,
+ clusterId,
+ () => {
+ val brokers = new
DescribeClusterResponseData.DescribeClusterBrokerCollection()
+
metadataCache.getAliveBrokerNodes(request.context.listenerName).foreach { node
=>
+ brokers.add(new DescribeClusterResponseData.DescribeClusterBroker().
+ setBrokerId(node.id).
+ setHost(node.host).
+ setPort(node.port).
+ setRack(node.rack))
+ }
+ brokers
+ },
+ () => {
+ metadataCache.getControllerId match {
+ case Some(value) =>
+ value match {
+ case ZkCachedControllerId (id) => id
+ case KRaftCachedControllerId (_) =>
metadataCache.getRandomAliveBrokerId.getOrElse(- 1)
Review Comment:
Hi @cmccabe,
Thanks for the PR! Sorry for revisiting this, but since this change allows
clients to communicate directly with controllers, wouldn't it make more sense
to return the active controller ID here instead of a random broker ID?
--
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
