Producer does not send anything unless stop() is called.

[ant0nk]

Describe the bug Continuously running application periodically calls AIOKafkaProducer.send(), there is a corresponding record in log - Sending (key=None value=... but nothing gets to Kafka actually unless stop() is called.

Expected behaviour Message gets to Kafka right after send() is called.

Environment (please complete the following information):

• aiokafka version (python -c "import aiokafka; print(aiokafka.__version__)"): 0.8.1
• kafka-python version (python -c "import kafka; print(kafka.__version__)"): 2.0.2
• Kafka Broker version (kafka-topics.sh --version): -
• Other information (Confluent Cloud version, etc.): -

Reproducible example

import asyncio
from time import sleep
from aiokafka import AIOKafkaProducer
async def main():
    producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
    await producer.start()
    await producer.send('topic', b'test')
    while True:
        sleep(10)

asyncio.run(main())

[vmaurin]

Kafka is optimized for throughput and the main idea to achieve that with fewer brokers is to delegate a lot of work to the clients (producer and consumer). More or less, you will have the following flow :

• the producer partitions messages, group them in a chunk and compress them
• the producer send the chunk of messages, compressed to the broker with a TCP message
• the broker gets the chunk, and append it to a file "as is" (ideally the broker doesn't run anykind of processing on the chunk, just a simple copy from the socket to the file)
• the consumer polls the broker, sees there is a new chunk, fetches it (usually from broker cache memory as it is a file just written)
• the consumer decompresses the chunk and retrieve individual messages

The behavior you are observing then could be consider normal, as the producer tries to batch as many message as possible (there are some producer options to control that). There is also a "send_and_wait" method for the app to wait for a batch to be send, but then it will tend to have one message per chunk, it is not very efficient, see https://aiokafka.readthedocs.io/en/stable/producer.html#message-buffering

[ant0nk]

@vmaurin The link says "By default, a new batch is sent immediately after the previous one (even if it’s not full)". As linger_ms is set to 0 by default I expected that messages should be sent immediately. Now I see that send() method can just wait forever if messages are scarce.

[vmaurin]

Maybe it is there is weird, on scarce traffic, a batch should still be send at some point (like the next time the sender coroutine execute, that should follow up as there is trigger when data is ready to be drained). Could you provide a more complete example ? (where you have your "stop()" call for example)

What is a bit tricky with send is that the coroutine itself is just about appending the message to the batch, so it should be quite fast/immediate, i.e await send(...) go next before anything will be send on the network. Then this coroutine return a future that will be completed when the message is actually sent, so somehow the code of "send_and_await" is a bit like

fut = await send(...)
await fut

[ant0nk]

@vmaurin example with stop():

import asyncio
from time import sleep
from aiokafka import AIOKafkaProducer
async def main():
    producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
    await producer.start()
    await producer.send('topic', b'test')
    sleep(30)
    await producer.stop()

asyncio.run(main())

Message gets to Kafka only after stop() is called. And the example in initial post does not send anything at all.

[vmaurin]

Thank you for the snippet @ant0nk With a time.sleep, you are blocking the process running the asyncio event loop, so the sender co routine won't be able to run. Could you try the same sample with an await asyncio.sleep(30) instead ?

[ant0nk]

@vmaurin So when program have infinite cycle it must have asyncio steps to send() to work. For example if I have program like this:

import asyncio
import os
from aiokafka import AIOKafkaProducer
async def main():
    producer = AIOKafkaProducer(bootstrap_servers='localhost:9092')
    await producer.start()
    while True:
        if os.path.exists('/path/to/trigger.txt'):
            await producer.send('topic', b'I saw trigger.txt')
            os.remove('/path/to/trigger.txt')
asyncio.run(main())

it will send first message only when file /path/to/trigger.txt will be created for second time.

[vmaurin]

@ant0nk Yes, this one is a common advice when using asyncio/coroutine in python (but also in similar thing like javascript) "Don't block the event loop". So if you need to lookup for your file to exist, you can either try to find a lib to do file IO in a coroutine way, or use a thread pool to run you IO blocking operation + sync primitive to combine both https://docs.python.org/3/library/asyncio-task.html#running-in-threads

[ant0nk]

Thanks, closing issue.