Return list of describe config responses instead of merging them

Author: jlandersen

kafka/admin/client.py

@@ -820,19 +820,10 @@ def describe_configs(self, config_resources, include_synonyms=False):
                 ))
         else:
             raise NotImplementedError(
-                "Support for DescribeConfigs v{} has not yet been added to KafkaAdminClient."
-                    .format(version))
+                "Support for DescribeConfigs v{} has not yet been added to KafkaAdminClient.".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].value)
-        response.resources = []
-
-        for future in futures:
-            response.resources.extend(future.value.resources)
-
-        return response
+        return [f.value for f in futures]
 
     @staticmethod
     def _convert_alter_config_resource_request(config_resource):

test/test_admin_integration.py

@@ -3,9 +3,10 @@
 from test.testutil import env_kafka_version
 
 from kafka.errors import NoError
-from kafka.admin import ACLFilter, ACLOperation, ACLPermissionType, ResourcePattern, ResourceType, ACL
-
+from kafka.admin import (
+    ACLFilter, ACLOperation, ACLPermissionType, ResourcePattern, ResourceType, ACL, ConfigResource, ConfigResourceType)
 
+    
 @pytest.mark.skipif(env_kafka_version() < (0, 11), reason="ACL features require broker >=0.11")
 def test_create_describe_delete_acls(kafka_admin_client):
     """Tests that we can add, list and remove ACLs
@@ -80,3 +81,45 @@ def test_create_describe_delete_acls(kafka_admin_client):
 
     assert error is NoError
     assert len(acls) == 0
+
+
+@pytest.mark.skipif(env_kafka_version() < (0, 11), reason="Describe config features require broker >=0.11")
+def test_describe_configs_broker_resource_returns_configs(simple_client, kafka_admin_client):
+    """Tests that describe config returns configs for broker
+    """
+    broker_id = simple_client.brokers[0].nodeId
+    configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
+
+    assert len(configs) == 1
+    assert configs[0].resources[0][2] == ConfigResourceType.BROKER
+    assert configs[0].resources[0][3] == str(broker_id)
+    assert len(configs[0].resources[0][4]) > 1
+
+
+@pytest.mark.skipif(env_kafka_version() < (0, 11), reason="Describe config features require broker >=0.11")
+def test_describe_configs_topic_resource_returns_configs(simple_client, kafka_admin_client):
+    topic = simple_client.topics[0]
+    configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.TOPIC, topic)])
+
+    assert len(configs) == 1
+    assert configs[0].resources[0][2] == ConfigResourceType.TOPIC
+    assert configs[0].resources[0][3] == topic
+    assert len(configs[0].resources[0][4]) > 1
+
+
+@pytest.mark.skipif(env_kafka_version() < (0, 11), reason="Describe config features require broker >=0.11")
+def test_describe_configs_mixed_resources_returns_configs(simple_client, kafka_admin_client):
+    topic = simple_client.topics[0]
+    broker_id = simple_client.brokers[0].nodeId
+    configs = kafka_admin_client.describe_configs([
+        ConfigResource(ConfigResourceType.TOPIC, topic),
+        ConfigResource(ConfigResourceType.BROKER, broker_id)])
+
+    assert len(configs) == 2
+
+    for config in configs:
+        assert (config.resources[0][2] == ConfigResourceType.TOPIC
+                and config.resources[0][3] == topic) or \
+               (config.resources[0][2] == ConfigResourceType.BROKER
+                and config.resources[0][3] == str(broker_id))
+        assert len(config.resources[0][4]) > 1