Fix describe config for multi-broker clusters

Currently all describe config requests are sent to "least loaded node". Requests for broker configs must, however, be sent to the specific broker, otherwise an error is returned. Only topic requests can be handled by any node.

This changes the logic to send all describe config requests to the specific broker.

Author: jlandersen

kafka/admin/client.py

@@ -5,6 +5,7 @@
 import logging
 import socket
 
+from . import ConfigResourceType
 from kafka.vendor import six
 
 from kafka.client_async import KafkaClient, selectors
@@ -484,28 +485,74 @@ def describe_configs(self, config_resources, include_synonyms=False):
             supported by all versions. Default: False.
         :return: Appropriate version of DescribeConfigsResponse class.
         """
+
+        # Break up requests by type - a broker config request must be sent to the specific broker.
+        # All other (currently just topic resources) can be sent to any broker.
+        broker_resources = []
+        topic_resources = []
+
+        for config_resource in config_resources:
+            if config_resource.resource_type == ConfigResourceType.BROKER:
+                broker_resources.append(self._convert_describe_config_resource_request(config_resource))
+            else:
+                topic_resources.append(self._convert_describe_config_resource_request(config_resource))
+
+        futures = []
         version = self._matching_api_version(DescribeConfigsRequest)
         if version == 0:
             if include_synonyms:
                 raise IncompatibleBrokerVersion(
                     "include_synonyms requires DescribeConfigsRequest >= v1, which is not supported by Kafka {}."
-                    .format(self.config['api_version']))
-            request = DescribeConfigsRequest[version](
-                resources=[self._convert_describe_config_resource_request(config_resource) for config_resource in config_resources]
-            )
+                        .format(self.config['api_version']))
+
+            if len(broker_resources) > 0:
+                for broker_resource in broker_resources:
+                    try:
+                        futures.append(self._send_request_to_node(
+                            int(broker_resource[1]),
+                            DescribeConfigsRequest[version](resources=[broker_resource])
+                        ))
+                    except ValueError:
+                        raise ValueError("Broker resource names must be an integer or a string represented integer")
+
+            if len(topic_resources) > 0:
+                futures.append(self._send_request_to_node(
+                    self._client.least_loaded_node(),
+                    DescribeConfigsRequest[version](resources=topic_resources)
+                ))
+
         elif version == 1:
-            request = DescribeConfigsRequest[version](
-                resources=[self._convert_describe_config_resource_request(config_resource) for config_resource in config_resources],
-                include_synonyms=include_synonyms
-            )
+            if len(broker_resources) > 0:
+                for broker_resource in broker_resources:
+                    try:
+                        futures.append(self._send_request_to_node(
+                            int(broker_resource[1]),
+                            DescribeConfigsRequest[version](
+                                resources=[broker_resource],
+                                include_synonyms=include_synonyms)
+                        ))
+                    except ValueError:
+                        raise ValueError("Broker resource names must be an integer or a string represented integer")
+
+            if len(topic_resources) > 0:
+                futures.append(self._send_request_to_node(
+                    self._client.least_loaded_node(),
+                    DescribeConfigsRequest[version](resources=topic_resources, include_synonyms=include_synonyms)
+                ))
         else:
             raise NotImplementedError(
                 "Support for DescribeConfigs v{} has not yet been added to KafkaAdminClient."
-                .format(version))
-        future = self._send_request_to_node(self._client.least_loaded_node(), request)
+                    .format(version))
+
+        self._wait_for_futures(futures)
+
+        # Use one of the results as the general response and add all other resources to it
+        response = copy.copy(futures[0])
+        response.resources = []
+
+        for future in futures:
+            response.resources.extend(future.value.resources)
 
-        self._wait_for_futures([future])
-        response = future.value
         return response
 
     @staticmethod

test/test_admin_integration.py

@@ -0,0 +1,66 @@
+import os
+
+from kafka import KafkaAdminClient
+from kafka.admin import ConfigResource, ConfigResourceType
+from test.fixtures import ZookeeperFixture, KafkaFixture
+from test.testutil import KafkaIntegrationTestCase, kafka_versions
+
+
+class TestAdminClientIntegration(KafkaIntegrationTestCase):
+    @classmethod
+    def setUpClass(cls):  # noqa
+        if not os.environ.get('KAFKA_VERSION'):
+            return
+
+        cls.zk = ZookeeperFixture.instance()
+        cls.server = KafkaFixture.instance(0, cls.zk)
+
+    @classmethod
+    def tearDownClass(cls):  # noqa
+        if not os.environ.get('KAFKA_VERSION'):
+            return
+
+        cls.server.close()
+        cls.zk.close()
+
+    def admin_client(self, **configs):
+        brokers = '%s:%d' % (self.server.host, self.server.port)
+        admin_client = KafkaAdminClient(
+            bootstrap_servers=brokers, **configs)
+        return admin_client
+
+    @kafka_versions('>=0.10.0')
+    def test_describe_configs_broker_resource_returns_configs(self):
+        broker_id = self.client.brokers[0].nodeId
+        configs = self.admin_client().describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
+
+        self.assertEqual(len(configs.resources), 1)
+        self.assertEqual(configs.resources[0][2], ConfigResourceType.BROKER)
+        self.assertEqual(configs.resources[0][3], str(broker_id))
+        self.assertGreater(len(configs.resources[0][4]), 1)
+
+    @kafka_versions('>=0.10.0')
+    def test_describe_configs_topic_resource_returns_configs(self):
+        topic = self.client.topics[0]
+        configs = self.admin_client().describe_configs([ConfigResource(ConfigResourceType.TOPIC, topic)])
+
+        self.assertEqual(len(configs.resources), 1)
+        self.assertEqual(configs.resources[0][2], ConfigResourceType.TOPIC)
+        self.assertEqual(configs.resources[0][3], topic)
+        self.assertGreater(len(configs.resources[0][4]), 1)
+
+    @kafka_versions('>=0.10.0')
+    def test_describe_configs_mixed_resources_returns_configs(self):
+        topic = self.client.topics[0]
+        broker_id = self.client.brokers[0].nodeId
+        configs = self.admin_client().describe_configs([
+            ConfigResource(ConfigResourceType.TOPIC, topic),
+            ConfigResource(ConfigResourceType.BROKER, broker_id)])
+
+        self.assertEqual(len(configs.resources), 2)
+
+        for resource in configs.resources:
+            self.assertTrue(
+                (configs.resources[0][2] == ConfigResourceType.TOPIC and configs.resources[0][3] == topic) or
+                (configs.resources[0][2] == ConfigResourceType.BROKER and configs.resources[0][3] == str(broker_id)))
+            self.assertGreater(len(resource[4]), 1)