Add context manager support to Consumers and producers

[lyytinen]

Would it make sense for the consumers and producers to implement the context management protocol?
So that one could write:

from kafka import KafkaProducer
with KafkaProducer(bootstrap_servers='localhost:1234') as producer:
    for _ in range(100):
        producer.send('foobar', b'some_message_bytes')

[tvoinarovskyi]

Could the producer be entered again? It's just that with DB's it's common to have a pool of connections and with protocol typically is used to acquire 1 connection. Would adding context support not cause confusion like:

global_producer = KafkaProducer(bootstrap_servers='localhost:1234')

def send_message_handler(self, request):
    with global_producer as conn:
        conn.send("foobar", b"some_message_bytes")

And from this example on, the Producer and Consumer are both long running instances, and only in scripts do you create/destroy it in 1 function. Mostly it would be initialized in startup hook and stopped in teardown or something.
At least those were my thoughts about not adding it for aiokafka...

[lyytinen]

Fair points. It's true that the use cases that would benefit from context support are marginal but it would still be nice to do those with ease. Not entirely sure about the added confusion also because by using "with" you already acknowledge that something is going to happen behind the curtain.

[dpkp]

fwiw, there is an old (unsupported) context manager in kafka.context that was designed to manage consumer offsets w/ SimpleConsumer.

[dpkp]

Another alternative would be:

import contextlib
from kafka import KafkaProducer

with contextlib.closing(KafkaProducer(bootstrap_servers='localhost:1234')) as producer:
    for _ in range(100):
        producer.send('foobar', b'some_message_bytes')