Repository: Parsely/pykafka
Pull request #623 (2016-12-05 19:12:40)

remove member_id from MemberAssignment
This pull request removes the `member_id` attribute from the `MemberAssignment` class. This makes the class match the `MemberAssignment` schema from the pykafka protocol documentation.


Patch:
diff --git a/pykafka/managedbalancedconsumer.py b/pykafka/managedbalancedconsumer.py
index bc4ce8a..6abd131 100644
--- a/pykafka/managedbalancedconsumer.py
+++ b/pykafka/managedbalancedconsumer.py
@@ -279,11 +279,12 @@ class ManagedBalancedConsumer(BalancedConsumer):
                 # generate partition assignments for each group member
                 # if this is not the leader, join_result.members will be empty
                 group_assignments = [
-                    MemberAssignment([
+                    (member_id,
+                     MemberAssignment([
                         (self._topic.name,
                          [p.id for p in self._decide_partitions(
-                          iterkeys(members), consumer_id=member_id)])
-                    ], member_id=member_id) for member_id in members]
+                          iterkeys(members), consumer_id=member_id)])])
+                    ) for member_id in members]
 
                 assignment = self._sync_group(group_assignments)
                 self._setup_internal_consumer(
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index 2541eba..19b60b7 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -1365,8 +1365,7 @@ class MemberAssignment(object):
             Partition => int32
         UserData => bytes
     """
-    def __init__(self, partition_assignment, member_id=None, version=1):
-        self.member_id = member_id
+    def __init__(self, partition_assignment, version=1):
         self.version = version
         self.partition_assignment = partition_assignment
 
@@ -1436,9 +1435,9 @@ class SyncGroupRequest(Request):
         # + len(member id) + member id + len(group assignment)
         size += 2 + len(self.member_id) + 4
         # group assignment tuples
-        for member_assignment in self.group_assignment:
+        for member_id, member_assignment in self.group_assignment:
             # + len(member id) + member id + len(member assignment) + member assignment
-            size += 2 + len(member_assignment.member_id) + 4 + len(member_assignment)
+            size += 2 + len(member_id) + 4 + len(member_assignment)
         return size
 
     @property
@@ -1460,12 +1459,11 @@ class SyncGroupRequest(Request):
                          self.generation_id, len(self.member_id), self.member_id,
                          len(self.group_assignment))
         offset += struct.calcsize(fmt)
-        for member_assignment in self.group_assignment:
+        for member_id, member_assignment in self.group_assignment:
             assignment_bytes = bytes(member_assignment.get_bytes())
-            fmt = '!h%dsi%ds' % (len(member_assignment.member_id), len(assignment_bytes))
-            struct.pack_into(fmt, output, offset, len(member_assignment.member_id),
-                             member_assignment.member_id, len(assignment_bytes),
-                             assignment_bytes)
+            fmt = '!h%dsi%ds' % (len(member_id), len(assignment_bytes))
+            struct.pack_into(fmt, output, offset, len(member_id), member_id,
+                             len(assignment_bytes), assignment_bytes)
             offset += struct.calcsize(fmt)
         return output
 


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 8117daa..738bfac 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -636,10 +636,10 @@ class TestGroupMembershipAPI(unittest2.TestCase):
         req = protocol.SyncGroupRequest(
             b'dummygroup', 1, b'testmember1',
             [
-                protocol.MemberAssignment([(b"mytopic1", [3, 5, 7, 9]),
-                                           (b"mytopic2", [3, 5, 7, 9])], member_id=b"a"),
-                protocol.MemberAssignment([(b"mytopic1", [2, 4, 6, 8]),
-                                           (b"mytopic2", [2, 4, 6, 8])], member_id=b"b")
+                (b"a", protocol.MemberAssignment([(b"mytopic1", [3, 5, 7, 9]),
+                                                  (b"mytopic2", [3, 5, 7, 9])])),
+                (b"b", protocol.MemberAssignment([(b"mytopic1", [2, 4, 6, 8]),
+                                                  (b"mytopic2", [2, 4, 6, 8])]))
             ])
         msg = req.get_bytes()
         self.assertEqual(