aiokafka concurrent processing

[lesnik512]

Hi! Thanks for this amazing framework.

We’re actively using FastStream with the aiokafka broker and are trying to understand how to improve subscriber throughput using concurrent processing.

In subscribers I see a few options:

1. max_workers > 1 (concurrent processing)
2. batch=True with max_records > 1 (batch processing)
3. using both together

Could you please clarify how offset commits / acknowledgements work in each case, especially with ack_policy=AckPolicy.ACK?

Concrete questions:

1. Batch mode: if I use batch=True, max_records=100, ack_policy=AckPolicy.ACK — is the entire batch acknowledged/committed only after the handler finishes successfully (i.e., no exception)?

2. Concurrency (max_workers > 1): with ack_policy=AckPolicy.ACK, does FastStream provide at-least-once processing semantics?

   • Specifically, are offsets committed only after the corresponding message is processed successfully?
   • If multiple messages from the same partition are processed concurrently and finish out of order, does FastStream ensure commits remain contiguous (i.e., it won’t commit past an unprocessed offset)?

Thanks in advance — a short explanation of the commit/ack strategy (and failure behavior) for these modes would really help us tune performance safely.

P.S. Sorry, if these questinos are already answered somewhere, but I couldn't find :(

[lesnik512]

Found issue where max_workers where added #2017

So max_workers>1 enables concurrent-between-partitions kafka subscriber, meaning concurrency limited to the number of partitions.

[lesnik512]

And another question is how batch size is defined? There are several arguments, max_records, max_poll_records and there is no info in docs https://faststream.ag2.ai/latest/kafka/Subscriber/batch_subscriber/

[lesnik512]

seems like max_records is for batch=True

and max_poll_records is from aiokafka and it's the maximum number of records returned in a single call to getmany() (from https://aiokafka.readthedocs.io/en/stable/api.html#consumer-class)

[lesnik512]

Wrote with my team package https://github.com/modern-python/faststream-concurrent-aiokafka

Concurrent message processing middleware for FastStream with aiokafka.

By default FastStream processes Kafka messages sequentially — one message at a time per subscriber. This library turns each incoming message into an asyncio task so multiple messages are handled concurrently, while keeping offset commits correct and shutdown graceful.

Now trying it in production