This is quadratic explosion due to the fact that a batch of messages is limited in size. If you have 10000 messages waiting in parallel and can only send 100 at once, then every time 100 messages are sent, all the 9900 waiting messages will be woken up all together, but 9800 will have to sleep back once the queue is full again.

There's actually no interest in parallel calls to send since they are just appended in a list sequentially. If you want better performances, try creating several producers, then you can send several requests at once (but you'll loose the order).

Hello @aure-olli and thank you for your response :)

If you want better performances, try creating several producers, then you can send several requests at once (but you'll loose the order).

I will try this solution since order is not really important to my usecase, thank you!

I will test this scenario, it maybe makes sense to find some way to handle it under the hood. It may even be less of a problem now, as we removed the recursion. As for a workaround you can also try to work it out using "batch" API for sending. As you handle the scheduling yourself it will not have such problems. See https://aiokafka.readthedocs.io/en/stable/producer.html#direct-batch-control

Is this fixed or still a problem?

@patkivikram Hi, can you reproduce the behavior, is this problematic for you?
I found the fix for the problem quite problematic to implement actually, as it would require a lot of rewriting. It's a rather specific use-case, so I did not work on it yet. Asyncio is constantly enhancing its structures so it may as well be not as bad in the newer versions too. If you have a specific use-case that is hitting the issue I can help you either work around it or prioritize the fix more.

@tvoinarovskyi I am seeing a bottleneck where my producer is writing to multiple topics and partitions and gets full even with a high batch-size. Should I use a different producer per topic/partition?

I reproduced it. I have an HTTP API, which receives data and pushes it to Kafka. 10-20 QPS concurrency can keep the CPU at 100%.

I "solved" this by putting a Semaphore before sending the request, looks like aiokafka cant handle having so many pending tasks in its buffer

This is the benchmark code:

import asyncio
from asyncio import as_completed
import time

from aiokafka import AIOKafkaProducer


async def send_msg(msg, limiter, producer):
    async with limiter:
        await producer.send_and_wait(value=msg, topic="my_topic",
                                     headers=[('x-datadog-trace-id', b'12593897844508091594'),
                                              ('x-datadog-parent-id', b'3332965923773247704'),
                                              ('x-datadog-sampling-priority', b'1')])


async def main():
    rate_limier = asyncio.Semaphore(150)
    async with AIOKafkaProducer(bootstrap_servers="localhost:9092", client_id="abc", acks=1) as producer:
        msgs = [b"qwjeoiqjwelkqndlkqweoiqwjeqliwjeoi12u3o1i2uwjeqlwiejqlwksdalkwnoiqweoi12u3" \
                b"iqwjlaksdlm1o23uo12ueolqwdjlakwejo12iu3oiejdlqakwsdnasdalksdalksdjalksdjalsdjal" \
                b"ksdjalsdja;ldsjaksdjaskje123u1023u12i3j12lk3j12k3123"
                for i in range(30_000)]
        tasks = [send_msg(msg, rate_limier, producer) for msg in msgs]

        start = time.time()
        sent = 0
        for fut in as_completed(tasks):
            await fut
            sent += 1
            print(f'\r{sent} | rate={sent/(time.time()-start):.2f} msgs/s', end='', flush=True)
        total_time = time.time() - start
        print(f'\n{total_time=:.2f} | rate={30_000/total_time:.2f} msgs/s')


if __name__ == '__main__':
    asyncio.run(main())

This is the output with the semaphore:

30000 | rate=13573.36 msgs/s
total_time=2.21 | rate=13573.01 msgs/s

This is without it:

5250 | rate=120.97 msgs/s

After 45 seconds everything timed out and crashed

The following code is my solution.

import asyncio

from loguru import logger
from indexpy import HTTPException

from .mq import get_kafka_client

push_queue: asyncio.Queue[
    tuple[asyncio.Event, tuple[str, bytes, bytes]]
] = asyncio.Queue(1000)


async def _real_push():
    while not push_queue.empty():
        event, (topic, value, key) = await push_queue.get()
        try:
            await get_kafka_client().send_and_wait(topic=topic, value=value, key=key)
        except Exception as e:
            logger.error(f"Failed to push message to Kafka: {e}")
        finally:
            event.set()


worker: asyncio.Task | None = None


async def push_message(topic: str, value: bytes, key: bytes) -> None:
    """
    Push a message to a Kafka topic.
    """
    event = asyncio.Event()
    await push_queue.put((event, (topic, value, key)))
    global worker
    if worker is None:
        worker = asyncio.create_task(_real_push())
    if worker.done():
        worker = asyncio.create_task(_real_push())
    await event.wait()

I "solved" this by putting a Semaphore before sending the request, looks like aiokafka cant handle having so many pending tasks in its buffer

This is the benchmark code:

import asyncio
from asyncio import as_completed
import time

from aiokafka import AIOKafkaProducer


async def send_msg(msg, limiter, producer):
    async with limiter:
        await producer.send_and_wait(value=msg, topic="my_topic",
                                     headers=[('x-datadog-trace-id', b'12593897844508091594'),
                                              ('x-datadog-parent-id', b'3332965923773247704'),
                                              ('x-datadog-sampling-priority', b'1')])


async def main():
    rate_limier = asyncio.Semaphore(150)
    async with AIOKafkaProducer(bootstrap_servers="localhost:9092", client_id="abc", acks=1) as producer:
        msgs = [b"qwjeoiqjwelkqndlkqweoiqwjeqliwjeoi12u3o1i2uwjeqlwiejqlwksdalkwnoiqweoi12u3" \
                b"iqwjlaksdlm1o23uo12ueolqwdjlakwejo12iu3oiejdlqakwsdnasdalksdalksdjalksdjalsdjal" \
                b"ksdjalsdja;ldsjaksdjaskje123u1023u12i3j12lk3j12k3123"
                for i in range(30_000)]
        tasks = [send_msg(msg, rate_limier, producer) for msg in msgs]

        start = time.time()
        sent = 0
        for fut in as_completed(tasks):
            await fut
            sent += 1
            print(f'\r{sent} | rate={sent/(time.time()-start):.2f} msgs/s', end='', flush=True)
        total_time = time.time() - start
        print(f'\n{total_time=:.2f} | rate={30_000/total_time:.2f} msgs/s')


if __name__ == '__main__':
    asyncio.run(main())

This is the output with the semaphore:

30000 | rate=13573.36 msgs/s
total_time=2.21 | rate=13573.01 msgs/s

This is without it:

5250 | rate=120.97 msgs/s

After 45 seconds everything timed out and crashed

Updating my previous solution, I recently hit a bottleneck with my previous "solution" when using a Semaphore, I instead wrote this function, and got MUCH faster results, x2 in the benchmark for example.

import asyncio
import random
from asyncio import as_completed
import time

from aiokafka import AIOKafkaProducer
from typing import List

async def write_msgs_to_kafka(kafka_conn: AIOKafkaProducer, msgs: List[bytes], topic: str):
    batches = [kafka_conn.create_batch()]
    for msg in msgs:
        if not batches[-1].append(key=None, value=msg, timestamp=None, headers=[('x-datadog-trace-id', b'12593897844508091594'),
                                              ('x-datadog-parent-id', b'3332965923773247704'),
                                              ('x-datadog-sampling-priority', b'1')]):
            batches.append(kafka_conn.create_batch())
            batches[-1].append(key=None, value=msg, timestamp=None)

    futs = await asyncio.gather(*[
        kafka_conn.send_batch(batch, topic=topic, partition=random.randint(0, 3)) for batch in batches
    ])

    await asyncio.gather(*futs)

async def main_batch():
    async with AIOKafkaProducer(bootstrap_servers="localhost:9092", client_id="abc", acks=1,
                                request_timeout_ms=40000 * 3,
                                metadata_max_age_ms=300000 * 3) as producer:
        chunks = []
        for i in range(30):
            chunks.append([b"qwjeoiqjwelkqndlkqweoiqwjeqliwjeoi12u3o1i2uwjeqlwiejqlwksdalkwnoiqweoi12u3" \
                           b"iqwjlaksdlm1o23uo12ueolqwdjlakwejo12iu3oiejdlqakwsdnasdalksdalksdjalksdjalsdjal" \
                           b"ksdjalsdja;ldsjaksdjaskje123u1023u12i3j12lk3j12k3123"
                           for i in range(1000)])

        start = time.time()
        await asyncio.gather(*[write_msgs_to_kafka(producer, chunk, 'my_topic2') for chunk in chunks])
        total_time = time.time() - start
        print(f'\n{total_time=:.2f} | rate={30_000 / total_time:.2f} msgs/s')

total_time=1.09 | rate=27631.94 msgs/s

Also, when removing the headers, the semaphore method outputs 17341 msgs/s, but the batch method 52570 msgs/s !