Repository: Parsely/pykafka (python)

Issue:
How to resolve InconsistentGroupProtocol
Hi, we are migrating some of our consumers to golang for increased performance but still need to have pykafka consumers in the same consumer group.

When trying to join the group I am getting InconsistentGroupProtocol exceptions until I stop the GO consumers, in which case pykafka consumers can join the group and start. The same happens the other way around if the pykafka consumers are already started and we start the GO consumers - the GO consumers will throw a similar error and work when we stop the python consumers.

We are using [sarama](https://github.com/Shopify/sarama) and [sarama-cluster](https://github.com/bsm/sarama-cluster) for our GO consumers.

I have tested this with pykafka 2.6.0.dev2.
We are using Kafka 0.10.0.1

Could you give any insight on different group protocols, and whether or not it is possible to change these? Am I right in thinking that there are currently 2 versions of the protocol?

I have used [these docs](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-GroupMembershipAPI) to try to gain an understanding of what is happening.

Pull request:
use the standard `range` assignment strategy
This pull request makes the assumption that the "rebalancing algorithm" laid out [here](https://kafka.apache.org/documentation/#impl_consumerrebalance) describes exactly the semantics of the `range` partition assignment strategy described [here](https://kafka.apache.org/documentation/#oldconsumerconfigs). If that's not the case, we'll need to adjust the partition assignment strategy to exactly match those semantics.

Fixes #655

Patch:
diff --git a/pykafka/balancedconsumer.py b/pykafka/balancedconsumer.py
index fe575de..d061ffd 100644
--- a/pykafka/balancedconsumer.py
+++ b/pykafka/balancedconsumer.py
@@ -415,7 +415,7 @@ class BalancedConsumer(object):
         """Decide which partitions belong to this consumer.
 
         Uses the consumer rebalancing algorithm described here
-        http://kafka.apache.org/documentation.html
+        https://kafka.apache.org/documentation/#impl_consumerrebalance
 
         It is very important that the participants array is sorted,
         since this algorithm runs on each consumer and indexes into the same
diff --git a/pykafka/cli/kafka_tools.py b/pykafka/cli/kafka_tools.py
index 13f9f32..d782a9d 100644
--- a/pykafka/cli/kafka_tools.py
+++ b/pykafka/cli/kafka_tools.py
@@ -25,7 +25,9 @@ def fetch_offsets(client, topic, offset):
     :type client:  :class:`pykafka.KafkaClient`
     :param topic:  Name of the topic.
     :type topic:  :class:`pykafka.topic.Topic`
-    :param offset: Offset to fetch. Can be earliest, latest or a datetime.
+    :param offset: Offset to reset to. Can be earliest, latest or a datetime.
+        Using a datetime will reset the offset to the latest message published
+        *before* the datetime.
     :type offset: :class:`pykafka.common.OffsetType` or
         :class:`datetime.datetime`
     :returns: {partition_id: :class:`pykafka.protocol.OffsetPartitionResponse`}
@@ -145,7 +147,7 @@ def print_consumer_lag(client, args):
 
 
 def print_offsets(client, args):
-    """Print offsets for a topic
+    """Print offsets for a topic/consumer group.
 
     NOTE: Time-based offset lookups are not precise, but are based on segment
           boundaries. If there is only one segment, as when Kafka has just
@@ -317,10 +319,10 @@ def _get_arg_parser():
     _add_topic(parser)
     _add_consumer_group(parser)
 
-    # Print Broker Offsets
+    # Print Offsets
     parser = subparsers.add_parser(
         'print_offsets',
-        help='Fetch broker offsets for a topic'
+        help='Fetch offsets for a topic/consumer group'
     )
     parser.set_defaults(func=print_offsets)
     _add_topic(parser)
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index 0877ff5..d0bac25 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -1253,7 +1253,7 @@ GroupMembershipProtocol = namedtuple(
 )
 
 
-ConsumerGroupProtocol = GroupMembershipProtocol(b"consumer", b"pykafkaassignmentstrategy",
+ConsumerGroupProtocol = GroupMembershipProtocol(b"consumer", b"range",
                                                 ConsumerGroupProtocolMetadata())
 
 


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 314cf99..b54916a 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -561,7 +561,7 @@ class TestGroupMembershipAPI(unittest2.TestCase):
         self.assertEqual(
             msg,
             bytearray(
-                b'\x00\x00\x00|\x00\x0b\x00\x00\x00\x00\x00\x00\x00\x07pykafka'  # header
+                b'\x00\x00\x00h\x00\x0b\x00\x00\x00\x00\x00\x00\x00\x07pykafka'  # header
                 b'\x00\n'  # len(groupid)
                     b'dummygroup'  # groupid
                 b'\x00\x00u0'  # session timeout
@@ -570,8 +570,8 @@ class TestGroupMembershipAPI(unittest2.TestCase):
                 b'\x00\x08'  # len(protocol type)
                     b'consumer'  # protocol type
                 b'\x00\x00\x00\x01'  # len(group protocols)
-                    b'\x00\x19'  # len(protocol name)
-                        b'pykafkaassignmentstrategy'  # protocol name
+                    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
             )
@@ -775,8 +775,8 @@ class TestAdministrativeAPI(unittest2.TestCase):
                         b'Stable'  # state
                     b'\x00\x08'  # len(protocol_type)
                         b'consumer'  # protocol_type
-                    b'\x00\x19'  # len(protocol)
-                        b'pykafkaassignmentstrategy'  # protocol
+                    b'\x00\x05'  # len(protocol)
+                        b'range'  # protocol
                     b'\x00\x00\x00\x01'  # len(members)
                         b'\x00,'  # len(member_id)
                             b'pykafka-d42426fb-c295-4cd9-b585-6dd79daf3afe'  # member_id
@@ -800,7 +800,7 @@ class TestAdministrativeAPI(unittest2.TestCase):
         self.assertEqual(group_response.group_id, b'testgroup')
         self.assertEqual(group_response.state, b'Stable')
         self.assertEqual(group_response.protocol_type, b'consumer')
-        self.assertEqual(group_response.protocol, b'pykafkaassignmentstrategy')
+        self.assertEqual(group_response.protocol, b'range')
         member_id = b'pykafka-d42426fb-c295-4cd9-b585-6dd79daf3afe'
         self.assertTrue(member_id in group_response.members)
         member = group_response.members[member_id]