Twisted python to read from kafka and write to elasticsearch
<p>I am new to Twisted, this is my first program.</p>

<p>I can not find a way to use the KafkaConsumer from the kafka-python library and use treq to trigger a post request to elasticsearch.</p>

<p>I could decompose the problem in small pieces:
Create an kafka consumer iterator and read data from it (the topic may be huge)</p>

<pre><code>def consumeKafka():
    consumer = KafkaConsumer(bootstrap_servers="kafka:9092", auto_offset_reset='earliest')
    consumer.subscribe(['kafkapipeline'])
    for v in consumer:
        v.value
</code></pre>

<p>post to elasticsearch using treq</p>

<pre><code>def post(self):
    d = treq.post('http://es:9200/pro/pr/', self.data)
    d.addCallbacks(lambda x: print(x), lambda x: print("error %s " % x))
</code></pre>

<p>start the reactor</p>

<pre><code>from twisted.internet import reactor
reactor.callWhenRunning(consumeKafka)
reactor.run()
</code></pre>

<p>Any idea how to make this work?</p>

<p>I don't use Kafka at all, so I'm not sure if this will work for you. Also, I'm assuming your having trouble running Kafka and treq at the same time. A generic way I deal with iterators in Twisted is to use <code>inlineCallbacks</code> to wait for a result and then do something with that result afterwards.</p>

<pre><code>from twisted.internet import defer

@defer.inlineCallbacks
def consumeKafka():
    consumer = KafkaConsumer(bootstrap_servers="kafka:9092", auto_offset_reset='earliest')
    consumer.subscribe(['kafkapipeline'])
    for v in consumer:
        value = yield v.value
        # do stuff with value
</code></pre>

<p>Then you can simply call this function and the reactor will take care of the rest. So your main section will look like this:</p>

<pre><code>consumeKafka()
reactor.run()
</code></pre>

<p>Note that the <code>consumeKafka()</code> function returns a <code>Deferred</code> so you add callbacks and errbacks as needed. Once you get comfortable with this model, take a look at <code>Cooperator</code> objects for more functionality.</p>

python asynchronous twisted