You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Concurrent message processing middleware for [FastStream](https://faststream.airt.ai/) with aiokafka.
4
8
5
-
By default FastStream processes Kafka messages sequentially. This library allows you to process multiple messages concurrently using asyncio tasks, with optional batch offset committing.
9
+
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.
By default aiokafka auto-commits offsets. If you manage commits manually, enable `enable_batch_commit=True` to have the library commit offsets in batches after each task completes:
74
+
### KafkaConcurrentProcessingMiddleware
50
75
51
-
```python
52
-
await initialize_concurrent_processing(
53
-
context=context,
54
-
concurrency_limit=20,
55
-
commit_batch_size=100,
56
-
commit_batch_timeout_sec=5,
57
-
enable_batch_commit=True,
58
-
)
59
-
```
76
+
A FastStream `BaseMiddleware` subclass. Add it to your broker to enable concurrent processing. It wraps each incoming message in an asyncio task submitted to `KafkaConcurrentHandler`.
60
77
61
-
With batch commit enabled, offsets are committed per partition at the highest completed offset in each batch.
78
+
### KafkaConcurrentHandler
62
79
63
-
## Consumer group filtering
80
+
The processing engine. Manages:
81
+
- An `asyncio.Semaphore` to enforce `concurrency_limit`
82
+
- A set of in-flight asyncio tasks
83
+
- A background observer that periodically discards stale completed tasks
84
+
- Signal handlers for graceful shutdown
64
85
65
-
When multiple consumer groups subscribe to the same topic, producers can tag messages with a `topic_group` header to direct them to a specific group. The middleware skips messages whose `topic_group` header doesn't match the consumer's group ID. Messages with no `topic_group` header are always processed.
86
+
### KafkaBatchCommitter
66
87
67
-
```python
68
-
# Producer side — send to a specific consumer group only
69
-
await broker.publish(
70
-
{"data": "..."},
71
-
topic="my-topic",
72
-
headers={"topic_group": "group-a"},
73
-
)
74
-
```
88
+
Runs as a background asyncio task. Receives `KafkaCommitTask` objects, waits for each task's asyncio future to complete, then commits the max offset per partition to Kafka. Batching is triggered by size or timeout. If the committer's task dies, `CommitterIsDeadError` is raised to callers.
|`concurrency_limit`|`10`| Max concurrent asyncio tasks (minimum: 1) |
100
+
|`commit_batch_size`|`10`| Max messages per commit batch |
101
+
|`commit_batch_timeout_sec`|`10.0`| Max seconds before flushing a batch |
102
+
103
+
Returns the `KafkaConcurrentHandler` instance.
104
+
105
+
### `stop_concurrent_processing(context)`
106
+
107
+
Flush pending commits, wait for in-flight tasks (up to 10 s), then stop the handler.
108
+
109
+
### `KafkaConcurrentProcessingMiddleware`
110
+
111
+
FastStream middleware class. Pass it to `KafkaBroker(middlewares=[...])` or `broker.add_middleware(...)`.
112
+
113
+
## How It Works
114
+
115
+
1.**Message dispatch**: On each incoming message, `consume_scope` calls `handle_task()`, which acquires a semaphore slot then fires the handler coroutine as a background `asyncio.Task`.
116
+
117
+
2.**Concurrency control**: The semaphore blocks new tasks when `concurrency_limit` is reached. The slot is released via a done-callback when the task finishes or fails.
118
+
119
+
3.**Offset committing**: Each dispatched task is paired with its Kafka offset and consumer reference and enqueued in `KafkaBatchCommitter`. Once the task completes, the committer groups offsets by partition and calls `consumer.commit(partitions_to_offsets)` with `offset + 1` (Kafka's "next offset to fetch" convention).
120
+
121
+
4.**Graceful shutdown**: `stop_concurrent_processing` sets the shutdown event, flushes the committer, cancels the observer task, and calls `asyncio.gather` with a 10-second timeout to wait for all in-flight tasks.
0 commit comments