Repository: Parsely/pykafka
Pull request #622 (2016-12-02 23:32:37)

parse protocol metadata from joingroupresponse
This pull request fixes a lingering TODO on `JoinGroupResponse` by parsing the returned `ProtocolMetadata` bytestring into a `ConsumerGroupProtocolMetadata` python object. This object is also usable in the administrative API.


Patch:
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index ebaaa56..2541eba 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -1205,10 +1205,10 @@ class ConsumerGroupProtocolMetadata(object):
             Topic => string
         UserData => bytes
     """
-    def __init__(self):
-        self.version = 0
-        self.topic_names = [b"dummytopic"]
-        self.user_data = b"testuserdata"
+    def __init__(self, version=0, topic_names=None, user_data=b"testuserdata"):
+        self.version = version
+        self.topic_names = topic_names or [b"dummytopic"]
+        self.user_data = user_data
 
     def __len__(self):
         # version + len(topic names)
@@ -1235,6 +1235,18 @@ class ConsumerGroupProtocolMetadata(object):
         offset += struct.calcsize(fmt)
         return output
 
+    @classmethod
+    def from_bytestring(cls, buff):
+        if len(buff) == 0:
+            return cls()
+        fmt = 'h [S] Y'
+        response = struct_helpers.unpack_from(fmt, buff, 0)
+
+        version = response[0]
+        topic_names = response[1]
+        user_data = response[2]
+        return cls(version, topic_names, user_data)
+
 
 GroupMembershipProtocol = namedtuple(
     'GroupMembershipProtocol', ['protocol_type', 'protocol_name', 'metadata']
@@ -1338,8 +1350,8 @@ class JoinGroupResponse(Response):
         self.group_protocol = response[2]
         self.leader_id = response[3]
         self.member_id = response[4]
-        # TODO - parse metadata bytestring into ConsumerGroupProtocolMetadata?
-        self.members = {_id: meta for _id, meta in response[5]}
+        self.members = {_id: ConsumerGroupProtocolMetadata.from_bytestring(meta)
+                        for _id, meta in response[5]}
 
 
 class MemberAssignment(object):


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 555419e..8117daa 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -597,12 +597,14 @@ class TestGroupMembershipAPI(unittest2.TestCase):
         )
         self.assertEqual(response.generation_id, 1)
         self.assertEqual(response.group_protocol, b'dummyassignmentstrategy')
-        self.assertEqual(response.leader_id,
-                         b'pykafka-b2361322-674c-4e26-9194-305962636e57')
-        self.assertEqual(response.member_id,
-                         b'pykafka-b2361322-674c-4e26-9194-305962636e57')
-        self.assertEqual(response.members,
-                         {b'pykafka-b2361322-674c-4e26-9194-305962636e57': b'\x00\x00\x00\x00\x00\x01\x00\ndummytopic\x00\x00\x00\x0ctestuserdata'})
+        member_id = b'pykafka-b2361322-674c-4e26-9194-305962636e57'
+        self.assertEqual(response.leader_id, member_id)
+        self.assertEqual(response.member_id, member_id)
+        self.assertTrue(member_id in response.members)
+        metadata = response.members[member_id]
+        self.assertEqual(metadata.version, 0)
+        self.assertEqual(metadata.topic_names, [b"dummytopic"])
+        self.assertEqual(metadata.user_data, b"testuserdata")
 
     def test_member_assignment_construction(self):
         assignment = protocol.MemberAssignment([(b"mytopic1", [3, 5, 7, 9]),