Repository: Parsely/pykafka
Pull request #707 (2017-08-10 21:13:02)

send topic name in protocol metadata
This pull request fixes a bug causing all consumer group metadata to return `dummytopic` (the pykafka default topic name) as the topic name in requests to the administrative api. 

Related to #706, #664


Patch:
diff --git a/pykafka/broker.py b/pykafka/broker.py
index 012b52d..c28b9fe 100644
--- a/pykafka/broker.py
+++ b/pykafka/broker.py
@@ -416,7 +416,7 @@ class Broker(object):
     #  Group Membership API  #
     ##########################
 
-    def join_group(self, connection_id, consumer_group, member_id):
+    def join_group(self, connection_id, consumer_group, member_id, topic_name):
         """Send a JoinGroupRequest
 
         :param connection_id: The unique identifier of the connection on which to make
@@ -426,11 +426,14 @@ class Broker(object):
         :type consumer_group: bytes
         :param member_id: The ID of the consumer joining the group
         :type member_id: bytes
+        :param topic_name: The name of the topic to which to connect, used in protocol
+            metadata
+        :type topic_name: str
         """
         handler = self._get_unique_req_handler(connection_id)
         if handler is None:
             raise SocketDisconnectedError
-        future = handler.request(JoinGroupRequest(consumer_group, member_id))
+        future = handler.request(JoinGroupRequest(consumer_group, member_id, topic_name))
         self._handler.sleep()
         return future.get(JoinGroupResponse)
 
diff --git a/pykafka/managedbalancedconsumer.py b/pykafka/managedbalancedconsumer.py
index 4c91bd5..a9cd2d2 100644
--- a/pykafka/managedbalancedconsumer.py
+++ b/pykafka/managedbalancedconsumer.py
@@ -344,7 +344,8 @@ class ManagedBalancedConsumer(BalancedConsumer):
         for i in range(self._cluster._max_connection_retries):
             join_result = self._group_coordinator.join_group(self._connection_id,
                                                              self._consumer_group,
-                                                             self._consumer_id)
+                                                             self._consumer_id,
+                                                             self._topic.name)
             if join_result.error_code == 0:
                 break
             log.info("Error code %d encountered during JoinGroupRequest for"
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index 66b6578..1dc31e5 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -1353,10 +1353,6 @@ GroupMembershipProtocol = namedtuple(
 )
 
 
-ConsumerGroupProtocol = GroupMembershipProtocol(b"consumer", b"range",
-                                                ConsumerGroupProtocolMetadata())
-
-
 class JoinGroupRequest(Request):
     """A group join request
 
@@ -1371,9 +1367,10 @@ class JoinGroupRequest(Request):
             ProtocolName => string
             ProtocolMetadata => bytes
     """
-    def __init__(self, group_id, member_id, session_timeout=30000):
+    def __init__(self, group_id, member_id, topic_name, session_timeout=30000):
         """Create a new group join request"""
-        self.protocol = ConsumerGroupProtocol
+        metadata = ConsumerGroupProtocolMetadata(topic_names=[topic_name])
+        self.protocol = GroupMembershipProtocol(b"consumer", b"range", metadata)
         self.group_id = group_id
         self.session_timeout = session_timeout
         self.member_id = member_id


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 8f1ef44..469fa44 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -590,7 +590,7 @@ class TestGroupMembershipAPI(unittest2.TestCase):
         )
 
     def test_join_group_request(self):
-        req = protocol.JoinGroupRequest(b'dummygroup', member_id=b'testmember')
+        req = protocol.JoinGroupRequest(b'dummygroup', b'testmember', b'abcdefghij')
         msg = req.get_bytes()
         self.assertEqual(
             msg,
@@ -607,7 +607,7 @@ class TestGroupMembershipAPI(unittest2.TestCase):
                     b'\x00\x05'  # len(protocol name)
                         b'range'  # protocol name
                     b'\x00\x00\x00"'  # len(protocol metadata)
-                        b'\x00\x00\x00\x00\x00\x01\x00\ndummytopic\x00\x00\x00\x0ctestuserdata'  # protocol metadata
+                        b'\x00\x00\x00\x00\x00\x01\x00\nabcdefghij\x00\x00\x00\x0ctestuserdata'  # protocol metadata
             )
         )