Include "start" in the try/finally in docs

Both producer and consumer classes have a "start" and "stop" method.

`start` initializes background co-routine, open connections and execute
various tasks that need to be cleaned up with the `stop` method.

But `stop` also hold a logic based on a boolean `_closed` that is
initialized to `False` at the end of the `__init__` method.

Thus, if one create a consumer or a producer and doesn't call `stop`
even if `start` was never call, the finalizer in `__del__` will send a
warning.

A correct pattern of calling `stop` in a try/finally block seems to be
then to do the `try` just after the `__init__` and not just after the
`start` as an error or a task cancellation in between would leave the
producer/consumer "open" and trigger the warning

fixes #938

Author: Vincent Maurin

README.rst

@@ -27,9 +27,9 @@ Example of AIOKafkaProducer usage:
 
     async def send_one():
         producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
-        # Get cluster layout and initial topic/partition leadership information
-        await producer.start()
         try:
+            # Get cluster layout and initial topic/partition leadership information
+            await producer.start()
             # Produce message
             await producer.send_and_wait("my_topic", b"Super message")
         finally:
@@ -58,9 +58,9 @@ Example of AIOKafkaConsumer usage:
             'my_topic', 'my_other_topic',
             bootstrap_servers='localhost:9092',
             group_id="my-group")
-        # Get cluster layout and join group `my-group`
-        await consumer.start()
         try:
+            # Get cluster layout and join group `my-group`
+            await consumer.start()
             # Consume messages
             async for msg in consumer:
                 print("consumed: ", msg.topic, msg.partition, msg.offset,

aiokafka/consumer/consumer.py

@@ -509,6 +509,7 @@ async def stop(self):
 
         * Commit last consumed message if autocommit enabled
         * Leave group if used Consumer Groups
+        * Close the underlying Kafka client
         """
         if self._closed:
             return

docs/consumer.rst

@@ -14,8 +14,8 @@ from a Kafka cluster. Most simple usage would be::
         "my_topic",
         bootstrap_servers='localhost:9092'
     )
-    await consumer.start()
     try:
+        await consumer.start()
         async for msg in consumer:
             print(
                 "{}:{:d}:{:d}: key={} value={} timestamp_ms={}".format(

docs/examples/local_state_consumer.rst

@@ -113,16 +113,15 @@ Local State consumer:
             auto_offset_reset="none",
             key_deserializer=lambda key: key.decode("utf-8") if key else "",
         )
-        await consumer.start()
 
         local_state = LocalState()
         listener = RebalanceListener(consumer, local_state)
-        consumer.subscribe(topics=["test"], listener=listener)
 
         save_task = asyncio.create_task(save_state_every_second(local_state))
 
         try:
-
+            await consumer.start()
+            consumer.subscribe(topics=["test"], listener=listener)
             while True:
                 try:
                     msg_set = await consumer.getmany(timeout_ms=1000)

docs/examples/ssl_consume_produce.rst

@@ -27,8 +27,8 @@ information.
             bootstrap_servers='localhost:9093',
             security_protocol="SSL", ssl_context=context)
 
-        await producer.start()
         try:
+            await producer.start()
             msg = await producer.send_and_wait(
                 'my_topic', b"Super Message", partition=0)
         finally:
@@ -37,8 +37,8 @@ information.
         consumer = AIOKafkaConsumer(
             "my_topic", bootstrap_servers='localhost:9093',
             security_protocol="SSL", ssl_context=context)
-        await consumer.start()
         try:
+            await consumer.start()
             consumer.seek(TopicPartition('my_topic', 0), msg.offset)
             fetch_msg = await consumer.getone()
         finally:

docs/examples/transaction_example.rst

@@ -50,15 +50,16 @@ process data and produce the resut to ``OUT_TOPIC`` in a transactional manner.
             group_id=GROUP_ID,
             isolation_level="read_committed"  # <-- This will filter aborted txn's
         )
-        await consumer.start()
 
         producer = AIOKafkaProducer(
             bootstrap_servers=BOOTSTRAP_SERVERS,
             transactional_id=TRANSACTIONAL_ID
         )
-        await producer.start()
 
         try:
+            await consumer.start()
+            await producer.start()
+
             while True:
                 msg_batch = await consumer.getmany(timeout_ms=POLL_TIMEOUT)
 

docs/index.rst

@@ -41,9 +41,9 @@ Here's a consumer example:
             'my_topic', 'my_other_topic',
             bootstrap_servers='localhost:9092',
             group_id="my-group")
-        # Get cluster layout and join group `my-group`
-        await consumer.start()
         try:
+            # Get cluster layout and join group `my-group`
+            await consumer.start()
             # Consume messages
             async for msg in consumer:
                 print("consumed: ", msg.topic, msg.partition, msg.offset,
@@ -71,9 +71,9 @@ Here's a producer example:
     async def send_one():
         producer = AIOKafkaProducer(
             bootstrap_servers='localhost:9092')
-        # Get cluster layout and initial topic/partition leadership information
-        await producer.start()
         try:
+            # Get cluster layout and initial topic/partition leadership information
+            await producer.start()
             # Produce message
             await producer.send_and_wait("my_topic", b"Super message")
         finally:

docs/producer.rst

@@ -9,8 +9,8 @@ Producer client
 to the Kafka cluster. Most simple usage would be::
 
     producer = aiokafka.AIOKafkaProducer(bootstrap_servers="localhost:9092")
-    await producer.start()
     try:
+        await producer.start()
         await producer.send_and_wait("my_topic", b"Super message")
     finally:
         await producer.stop()
@@ -103,8 +103,8 @@ by passing the parameter ``enable_idempotence=True`` to :class:`~.AIOKafkaProduc
     producer = aiokafka.AIOKafkaProducer(
         bootstrap_servers='localhost:9092',
         enable_idempotence=True)
-    await producer.start()
     try:
+        await producer.start()
         await producer.send_and_wait("my_topic", b"Super message")
     finally:
         await producer.stop()
@@ -134,8 +134,8 @@ attendant APIs, you must set the ``transactional_id`` configuration property::
     producer = aiokafka.AIOKafkaProducer(
         bootstrap_servers='localhost:9092',
         transactional_id="transactional_test")
-    await producer.start()
     try:
+        await producer.start()
         async with producer.transaction():
             res = await producer.send_and_wait(
                 "test-topic", b"Super transactional message")

examples/local_state_consumer.py

@@ -94,15 +94,15 @@ async def consume():
         auto_offset_reset="none",
         key_deserializer=lambda key: key.decode("utf-8") if key else "",
     )
-    await consumer.start()
 
     local_state = LocalState()
     listener = RebalanceListener(consumer, local_state)
-    consumer.subscribe(topics=["test"], listener=listener)
 
     save_task = asyncio.create_task(save_state_every_second(local_state))
 
     try:
+        await consumer.start()
+        consumer.subscribe(topics=["test"], listener=listener)
 
         while True:
             try:

examples/ssl_consume_produce.py

@@ -17,8 +17,8 @@ async def produce_and_consume():
         bootstrap_servers='localhost:9093',
         security_protocol="SSL", ssl_context=context)
 
-    await producer.start()
     try:
+        await producer.start()
         msg = await producer.send_and_wait(
             'my_topic', b"Super Message", partition=0)
     finally:
@@ -27,8 +27,8 @@ async def produce_and_consume():
     consumer = AIOKafkaConsumer(
         "my_topic", bootstrap_servers='localhost:9093',
         security_protocol="SSL", ssl_context=context)
-    await consumer.start()
     try:
+        await consumer.start()
         consumer.seek(TopicPartition('my_topic', 0), msg.offset)
         fetch_msg = await consumer.getone()
     finally: