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