How to properly consume kafka messages concurrently

[fernando-torres-blip-ai]

I need to scale up my kafka processing which is basically multiple asynchronous IO operations (requests). I understand that i can use multiple workers, but this is an leads to an uncessary usage of resource. I wonder if it is possible to make better usage of asyncio with faststream.

If im not doing something wrong, the subscriber below will process 1 message at time.

@router.subscriber("data-envelopes", group_id='my-group', auto_commit=False)
async def on_new_envelope(envelope: MessageEnvelope, msg: KafkaMessage):
    print(f'Received Envelope !')
    await asyncio.sleep(5)
    print('Finished Envelope !')

Log:

Received Envelope !
Finished Envelope !
Received Envelope !
Finished Envelope !
Received Envelope !
Finished Envelope !

I wish something just like fastapi's endpoints works. The below view will accepet multiple requests concurrently.

@router.post('/publish/envelope')
async def publish_envelope(envelope: MessageEnvelope):
    print('received publish')
    await router.publisher("data-envelopes").publish(envelope)
    await asyncio.sleep(10)
    print('finished publish')

I made the implementation below, but it leads to missing the manual Ack controllablility.

router = KafkaRouter(...)

async def process(envelope: MessageEnvelope, msg: KafkaMessage):
    print('Processing Envelope !')
    await asyncio.sleep(5)
    print('Finished Envelope !')
    await msg.nack() # <- this does not work


@router.subscriber("data-envelopes", group_id='my-group', auto_commit=False)
async def on_new_envelope(envelope: MessageEnvelope, msg: KafkaMessage):
    print(f'Received Envelope !')
    asyncio.create_task(process(envelope, msg))

Which should be the correct implementation to work with async concurrency and kafka ?

[Lancetnik]

Unfortunately, Kafka doesn't support general concurrent message consuming. You should commit offset one by one, in the concurrent case you can't recognize the correct offset position if any error occured. So, you should scale Kafka consumers by partitions - single consumer per partition.
But, if you are OK with "at most once" delivery and use auto_commit=True we can provide you with a concurrent messages processing via asyncio tasks. This feature is WIP now: #1912

[lesnik512]

Hi) Recently we've built solution https://github.com/modern-python/faststream-concurrent-aiokafka
It guarantees "at least once" processing. The trade-off is that processing order is not guaranteed

[MaximMukhametov]

Hi,

The Kafka subscriber docs (e.g. Basic Subscriber – Concurrent processing, v0.7) describe concurrent processing for max_workers > 1 only for:

• AckPolicy.ACK_FIRST (concurrent processing, at-most-once), and
• AckPolicy.NACK_ON_ERROR (concurrent between partitions, sequential within a partition, at-least-once).

My questions:
Why is AckPolicy.ACK not mentioned in that section?
Is ACK + max_workers > 1 unsupported, discouraged, equivalent to one of the two documented modes, or simply missing from the docs?
It would help to document explicitly what happens (subscriber type, ordering guarantees, commit timing, failure behavior) for ACK + max_workers > 1, if that combination is valid.

If middleware converts failures into “successful” completion (catch exception, perform side effects, then return without re-raising), AckPolicy.NACK_ON_ERROR’s error path would not run for those failures.
Is relying on this pattern together with NACK_ON_ERROR + max_workers > 1 a reasonable/supported way to get the documented concurrent-between-partitions behavior while still committing offsets after handled failures—without using AckPolicy.ACK_FIRST’s at-most-once semantics?
Or are there pitfalls (e.g. ordering, commit invariants, interaction with acknowledgement middleware) that make this fragile or unintended?

My goal is to consume asynchronously from a topic with 8 partitions using one consumer (processing messages one by one within each partition, but concurrently across partitions) using AckPolicy.ACK