Adding namedtuples for DescribeConsumerGroup response; Adding Serialization of MemberData and MemberAssignment for the response

Author: Apurva Telang

kafka/admin/client.py

@@ -1,9 +1,10 @@
 from __future__ import absolute_import
 
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 import copy
 import logging
 import socket
+from itertools import izip
 
 from . import ConfigResourceType
 from kafka.vendor import six
@@ -17,9 +18,11 @@
 from kafka.protocol.admin import (
     CreateTopicsRequest, DeleteTopicsRequest, DescribeConfigsRequest, AlterConfigsRequest, CreatePartitionsRequest,
     ListGroupsRequest, DescribeGroupsRequest, DescribeAclsRequest, CreateAclsRequest, DeleteAclsRequest)
+from kafka.protocol.types import Array
+from kafka.coordinator.protocol import ConsumerProtocolMemberMetadata, ConsumerProtocolMemberAssignment, ConsumerProtocol
 from kafka.protocol.commit import GroupCoordinatorRequest, OffsetFetchRequest
 from kafka.protocol.metadata import MetadataRequest
-from kafka.structs import TopicPartition, OffsetAndMetadata
+from kafka.structs import TopicPartition, OffsetAndMetadata, MemberInformation, GroupInformation
 from kafka.admin.acl_resource import ACLOperation, ACLPermissionType, ACLFilter, ACL, ResourcePattern, ResourceType, \
     ACLResourcePatternType
 from kafka.version import __version__
@@ -1000,22 +1003,47 @@ def _describe_consumer_groups_process_response(self, response):
         """Process a DescribeGroupsResponse into a group description."""
         if response.API_VERSION <= 3:
             assert len(response.groups) == 1
-            # TODO need to implement converting the response tuple into
-            # a more accessible interface like a namedtuple and then stop
-            # hardcoding tuple indices here. Several Java examples,
-            # including KafkaAdminClient.java
-            group_description = response.groups[0]
-            error_code = group_description[0]
+            for response_field, response_name in izip(response.SCHEMA.fields, response.SCHEMA.names):
+                if type(response_field) == Array:
+                    described_groups = response.__dict__[response_name]
+                    described_groups_field_schema = response_field.array_of
+                    for described_group  in described_groups:
+                        described_group_information_list = []
+                        is_consumer_protocol_type = False
+                        for (described_group_information, group_information_name, group_information_field) in izip(described_group, described_groups_field_schema.names, described_groups_field_schema.fields):
+                            if group_information_name == 'protocol_type':
+                                protocol_type = described_group_information
+                                is_consumer_protocol_type = (protocol_type == ConsumerProtocol.PROTOCOL_TYPE or not protocol_type)
+                            if type(group_information_field) == Array:
+                                member_information_list = []
+                                member_schema = group_information_field.array_of
+                                for members in described_group_information:
+                                    member_information = []
+                                    for (member, member_field, member_name)  in izip(members, member_schema.fields, member_schema.names):
+                                        if member_name == 'member_metadata' and is_consumer_protocol_type:
+                                            member_information.append(ConsumerProtocolMemberMetadata.decode(member))
+                                        elif member_name == 'member_assignment' and is_consumer_protocol_type:
+                                            member_information.append(ConsumerProtocolMemberAssignment.decode(member))
+                                        else:
+                                            member_information.append(member)
+                                    else:
+                                        member_info_tuple = MemberInformation._make(member_information)
+                                        member_information_list.append(member_info_tuple)
+                                else:
+                                    described_group_information_list.append(member_information_list)
+                            else:
+                                described_group_information_list.append(described_group_information)
+                        else:
+                            if response.API_VERSION <=2:
+                                described_group_information_list.append([])
+                            group_description = GroupInformation._make(described_group_information_list)
+            error_code = group_description.error_code
             error_type = Errors.for_code(error_code)
             # Java has the note: KAFKA-6789, we can retry based on the error code
             if error_type is not Errors.NoError:
                 raise error_type(
                     "DescribeGroupsResponse failed with response '{}'."
                     .format(response))
-            # TODO Java checks the group protocol type, and if consumer
-            # (ConsumerProtocol.PROTOCOL_TYPE) or empty string, it decodes
-            # the members' partition assignments... that hasn't yet been
-            # implemented here so just return the raw struct results
         else:
             raise NotImplementedError(
                 "Support for DescribeGroupsResponse_v{} has not yet been added to KafkaAdminClient."

kafka/structs.py

@@ -20,6 +20,11 @@
 OffsetAndTimestamp = namedtuple("OffsetAndTimestamp",
     ["offset", "timestamp"])
 
+MemberInformation = namedtuple("MemberInformation",
+    ["member_id", "client_id", "client_host", "member_metadata", "member_assignment"])
+
+GroupInformation = namedtuple("GroupInformation",
+    ["error_code", "group", "state", "protocol_type", "protocol", "members", "authorized_operations"])
 
 # Define retry policy for async producer
 # Limit value: int >= 0, 0 means no retries

test/test_admin_integration.py

@@ -2,7 +2,7 @@
 
 from test.testutil import env_kafka_version
 
-from kafka.errors import NoError
+from kafka.errors import (NoError, GroupCoordinatorNotAvailableError)
 from kafka.admin import (
     ACLFilter, ACLOperation, ACLPermissionType, ResourcePattern, ResourceType, ACL, ConfigResource, ConfigResourceType)
 
@@ -138,3 +138,20 @@ def test_describe_configs_invalid_broker_id_raises(kafka_admin_client):
 
     with pytest.raises(ValueError):
         configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
+
+@pytest.mark.skipif(env_kafka_version() < (0, 11), reason='Describe consumer group requires broker >=0.11')
+def test_describe_consumer_group_does_not_exist(kafka_admin_client):
+    """Tests that the describe consumer group call fails if the group coordinator is not available
+    """
+    with pytest.raises(GroupCoordinatorNotAvailableError):
+        group_description = kafka_admin_client.describe_consumer_groups(['test'])
+
+@pytest.mark.skipif(env_kafka_version() < (0, 11), reason='Describe consumer group requires broker >=0.11')
+def test_describe_consumer_group_exists(kafka_admin_client, kafka_consumer_factory, topic):
+    """Tests that the describe consumer group call returns valid consumer group information
+    """
+    consumer = kafka_consumer_factory(group_id='testgrp', auto_offset_reset='earliest')
+    consumer.poll(timeout_ms=20)
+    output = kafka_admin_client.describe_consumer_groups(['testgrp'])
+    assert output[0].group == 'testgrp'
+    assert output[0].members[0].member_metadata.subscription[0] == topic