Gevent implementation of kafka-python

[hmahmood]

This adds a 'green' submodule to kafka-python, which uses the gevent python library (www.gevent.org) constructs for certain classes. The only classes I ported over right now are Producer, SimpleProducer, KeyedProducer, KafkaConnection and KafkaClient.

To use the submodule, simply do replace

import kafka

with

import kafka.green as kafka

Comments welcome!

[quard8]

I have problem with this pull request.

Traceback (most recent call last):
  File "groot.py", line 24, in <module>
    producer = SimpleProducer(kafka, req_acks=SimpleProducer.ACK_NOT_REQUIRED, async=True)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/producer.py", line 196, in __init__
    batch_send_every_t)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/producer.py", line 120, in __init__
    self._setup_async(batch_send_every_n, batch_send_every_t)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/green/producer.py", line 13, in _setup_async
    self.client.copy(),
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/green/client.py", line 14, in copy
    return _KafkaClient(self.hosts, self.client_id, self.timeout)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/green/client.py", line 9, in __init__
    super(_KafkaClient, self).__init__(hosts=hosts, client_id=client_id, timeout=timeout)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/client.py", line 32, in __init__
    self.hosts = collect_hosts(hosts)
  File "/usr/local/Cellar/python/2.7.5/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/kafka/conn.py", line 28, in collect_hosts
    res = host_port.split(':')
AttributeError: 'tuple' object has no attribute 'split'

from kafka.green import KafkaClient, SimpleProducer
kafka = KafkaClient("localhost:9092")
producer = SimpleProducer(kafka, req_acks=SimpleProducer.ACK_NOT_REQUIRED, async=True)

Basically because of two calls of conn.collect_hosts func.

[hmahmood]

Should be fixed in the latest commit.

[dpkp]

probably should have some unit tests. also assume we would not add gevent to install requirements (but probably should be added to test requirements). and also think an addition to README is warranted if this gets merged.

[dpkp]

also -- there are currently some merge conflicts. please rebase and resolve

[hmahmood]

I am going to withdraw this request in favor of just using the gevent library's monkey patch feature. That just seems simpler. The copy() methods on KafkaConnection and KafkaClient don't work with gevent, but I will submit a patch for that shortly.