Simple consumer fails when consumer_group specified

[cscheffler]

pykafka version: 2.0.0

Behaviour: Cannot create a simple consumer with a consumer group id. You can create a simple consumer if consumer_group is not specified (but then you cannot call commit_offsets() – which is expected behaviour).

To reproduce:

from pykafka import KafkaClient
client = KafkaClient(hosts=HOSTS)
topic = client.topics[TOPIC]
consumer = topic.get_simple_consumer(consumer_group='some_identifier')

Exception:

Traceback (most recent call last):
  File "test.py", line 12, in <module>
    consumer = topic.get_simple_consumer(consumer_group='some_identifier')
  File "/home/ubuntu/carl/kafka-tests/venv/local/lib/python2.7/site-packages/pykafka/topic.py", line 159, in get_simple_consumer
    consumer_group=consumer_group, **kwargs)
  File "/home/ubuntu/carl/kafka-tests/venv/local/lib/python2.7/site-packages/pykafka/simpleconsumer.py", line 157, in __init__
    self._discover_offset_manager()
  File "/home/ubuntu/carl/kafka-tests/venv/local/lib/python2.7/site-packages/pykafka/simpleconsumer.py", line 232, in _discover_offset_manager
    self._offset_manager = self._cluster.get_offset_manager(self._consumer_group)
  File "/home/ubuntu/carl/kafka-tests/venv/local/lib/python2.7/site-packages/pykafka/cluster.py", line 283, in get_offset_manager
    res = future.get(ConsumerMetadataResponse)
  File "/home/ubuntu/carl/kafka-tests/venv/local/lib/python2.7/site-packages/pykafka/handlers.py", line 60, in get
     raise self.error
pykafka.exceptions.SocketDisconnectedError

[yungchin]

Hi, thanks for your bug reports today (I've just pushed proposed fixes for the other two). This one I cannot reproduce. Could it be that you're running a kafka cluster that doesn't permit broker-based offset storage?

(If that's the issue, we'd ideally need better error reporting for that case in pykafka.)

[cscheffler]

Hi, thanks for testing and responding so quickly.

Ah, I see. We are still on Kafka 0.8.1.1, which (I think) means that we are committing offsets via Zookeeper rather than using the Kafka brokers directly. Does pykafka require Kafka 0.8.2?

[yungchin]

I'm afraid so - there's no code in place to do zookeeper-based offset storage.

For later reference: #196 is the same issue, and might contain some clues on how we'd detect this situation and put in a helpful exception message.

[cscheffler]

Ok, thanks for the update. I'll see if we can upgrade Kafka.

Perhaps add the dependency on Kafka >= 0.8.2 to the README, or other part of the documentation? I haven't noticed it anywhere, but apologies if it is the docs and I missed it.

[emmettbutler]

My guess is that Kafka doesn't respond with anything if it gets an api key it doesn't understand, which would be the case when 0.8.1 is sent a ConsumerMetadataRequest. I imagine that catching SocketDisconnectedError here and raising a helpful error would probably be enough.