[LI-HOTFIX] Add list federated topic znodes rpc

clients/src/main/java/org/apache/kafka/clients/admin/Admin.java

@@ -319,14 +319,15 @@ default ListFederatedTopicZnodesResult listFederatedTopicZnodes() {
     }
 
     /**
-     * List federated topic znodes match given topic names
+     * List federated topic znodes match given topic names; if empty list passed, all existing federated topic znodes
+     * will be listed
      * @param federatedTopics topic names
      * @param options the options to use when list federated topic znodes
      * @return empty list if the given topic names' znode don't exist; otherwise return the federated topics formatted
      *         /namespace/topic
      */
     ListFederatedTopicZnodesResult listFederatedTopicZnodes(List<String> federatedTopics,
-        ListFederatedTopicZnodesOptions options);
+                                                            ListFederatedTopicZnodesOptions options);
 
     /**
      * Describe some topics in the cluster, with the default options.

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2034,7 +2034,7 @@ void handleFailure(Throwable throwable) {
 
     @Override
     public ListFederatedTopicZnodesResult listFederatedTopicZnodes(List<String> federatedTopics,
-        ListFederatedTopicZnodesOptions options) {
+                                                                   ListFederatedTopicZnodesOptions options) {
         final KafkaFutureImpl<List<String>> federatedTopicZnodesListingFuture = new KafkaFutureImpl<>();
         List<LiListFederatedTopicZnodesRequestData.FederatedTopics> topicsRequested = new ArrayList<>();
         federatedTopics.forEach(topic ->

clients/src/main/resources/common/message/LiListFederatedTopicZnodesRequest.json

@@ -26,4 +26,4 @@
         {"name": "Name", "type": "string", "versions": "0+", "about": "The topic name"}
       ]}
   ]
-}
+}

clients/src/main/resources/common/message/LiListFederatedTopicZnodesResponse.json

@@ -26,4 +26,4 @@
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." }
   ]
-}
+}

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -3820,26 +3820,37 @@ class KafkaApis(val requestChannel: RequestChannel,
     val listfederatedTopicZnodesRequest = request.body[LiListFederatedTopicZnodesRequest]
     val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldNeverReceive(request))
     val requestedTopics = listfederatedTopicZnodesRequest.data().topics()
+    val hasClusterAuthorization = authHelper.authorize(request.context, DESCRIBE, CLUSTER, CLUSTER_NAME)
 
     try {
-      if (requestedTopics.isEmpty || requestedTopics.size() == 0) {
+      if (requestedTopics.isEmpty) {
         // if empty list passed, list all existing federated topics
         val allFederatedTopicZnodes = zkSupport.zkClient.getAllFederatedTopics.toList.asJava
 
-        requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-          new LiListFederatedTopicZnodesResponse(
-            new LiListFederatedTopicZnodesResponseData().setTopics(allFederatedTopicZnodes)
-              .setThrottleTimeMs(requestThrottleMs), listfederatedTopicZnodesRequest.version()
+        // either return all federated topics or none, if cluster doesn't allow describe operation
+        if (hasClusterAuthorization) {
+          requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+            new LiListFederatedTopicZnodesResponse(
+              new LiListFederatedTopicZnodesResponseData().setTopics(allFederatedTopicZnodes)
+                .setThrottleTimeMs(requestThrottleMs), listfederatedTopicZnodesRequest.version()
+            )
           )
-        )
+        } else {
+          requestHelper.sendResponseExemptThrottle(request,
+            listfederatedTopicZnodesRequest.getErrorResponse(
+              new ClusterAuthorizationException("List all federated topic znodes operation not allowed"))
+          )
+        }
       } else {
         // if non-empty list passed, only list znode values for the given topics
         val foundFederatedTopicZnodes = mutable.Set[String]()
 
         requestedTopics.forEach(topic => {
-          val curFederatedTopicZnode = zkSupport.zkClient.getFederatedTopic(topic.name())
-          if (curFederatedTopicZnode != null) {
-            foundFederatedTopicZnodes.add(curFederatedTopicZnode)
+          if (hasClusterAuthorization || authHelper.authorize(request.context, DESCRIBE, TOPIC, topic.name())) {
+            val curFederatedTopicZnode = zkSupport.zkClient.getFederatedTopic(topic.name())
+            if (curFederatedTopicZnode != null) {
+              foundFederatedTopicZnodes.add(curFederatedTopicZnode)
+            }
           }
         })