Metadata-Version: 2.4
Name: faststream-concurrent-aiokafka
Version: 0.6.5
Summary: Concurrent message-processing middleware for FastStream + aiokafka
Keywords: faststream,kafka,aiokafka,concurrency,middleware,messaging,asyncio,python
Author: Artur Shiriev
Author-email: Artur Shiriev <me@shiriev.ru>
License-Expression: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Classifier: Topic :: Software Development :: Libraries
Requires-Dist: faststream[kafka]>=0.7.1,<0.8
Requires-Python: >=3.11, <4
Project-URL: Homepage, https://modern-python.org
Project-URL: Repository, https://github.com/modern-python/faststream-concurrent-aiokafka
Project-URL: Issues, https://github.com/modern-python/faststream-concurrent-aiokafka/issues
Project-URL: Changelog, https://github.com/modern-python/faststream-concurrent-aiokafka/releases
Description-Content-Type: text/markdown

<p align="center">
  <picture>
    <source media="(prefers-color-scheme: dark)"  srcset="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/faststream-concurrent-aiokafka/lockup-dark.svg">
    <source media="(prefers-color-scheme: light)" srcset="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/faststream-concurrent-aiokafka/lockup-light.svg">
    <img alt="faststream-concurrent-aiokafka" src="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/faststream-concurrent-aiokafka/lockup.png" width="420">
  </picture>
</p>

[![PyPI version](https://img.shields.io/pypi/v/faststream-concurrent-aiokafka.svg)](https://pypi.org/project/faststream-concurrent-aiokafka/)
[![Supported Python versions](https://img.shields.io/pypi/pyversions/faststream-concurrent-aiokafka.svg)](https://pypi.org/project/faststream-concurrent-aiokafka/)
[![Downloads](https://static.pepy.tech/badge/faststream-concurrent-aiokafka/month)](https://pepy.tech/projects/faststream-concurrent-aiokafka)
[![Coverage](https://img.shields.io/badge/coverage-100%25-brightgreen.svg)](https://github.com/modern-python/faststream-concurrent-aiokafka/actions/workflows/ci.yml)
[![CI](https://github.com/modern-python/faststream-concurrent-aiokafka/actions/workflows/ci.yml/badge.svg)](https://github.com/modern-python/faststream-concurrent-aiokafka/actions/workflows/ci.yml)
[![License](https://img.shields.io/github/license/modern-python/faststream-concurrent-aiokafka.svg)](https://github.com/modern-python/faststream-concurrent-aiokafka/blob/main/LICENSE)
[![GitHub stars](https://img.shields.io/github/stars/modern-python/faststream-concurrent-aiokafka)](https://github.com/modern-python/faststream-concurrent-aiokafka/stargazers)
[![Context7](https://img.shields.io/badge/Context7-docs-blue)](https://context7.com/modern-python/faststream-concurrent-aiokafka)
[![uv](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/uv/main/assets/badge/v0.json)](https://github.com/astral-sh/uv)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![ty](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ty/main/assets/badge/v0.json)](https://github.com/astral-sh/ty)

Concurrent message processing middleware for [FastStream](https://faststream.airt.ai/) 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.

## Features

- Concurrent message processing via asyncio tasks
- Configurable concurrency limit (semaphore-based)
- Batch offset committing per partition after each task completes
- Rebalance-safe: pending offsets are flushed on partition revocation via `ConsumerRebalanceListener`
- Fast shutdown: cancels in-flight tasks; uncommitted offsets are redelivered on restart (at-least-once)
- Signal handling owned by your lifespan / process manager — this lib does not register SIGTERM/SIGINT handlers
- Handler exceptions are logged but do not crash the consumer
- Health check helper to probe handler status from a `ContextRepo`

## Installation

```bash
pip install faststream-concurrent-aiokafka
```

## Quick Start

`ack_policy=AckPolicy.MANUAL` is **required** on every concurrent subscriber — the middleware enforces this at runtime.
Without it, FastStream would commit offsets before processing tasks complete, causing silent message loss on crash.
Subscribers on any other ack policy — `ACK_FIRST`, `ACK`, `REJECT_ON_ERROR`, `NACK_ON_ERROR` — are passed through without concurrent processing,
behaving exactly as they would if this middleware were not registered. That keeps a single broker-level registration safe across a mix of subscribers.

> **`AsgiFastStream` note**: its lifespan receives an app-level `ContextRepo` separate from `broker.context`. Pass `broker.context` explicitly instead of the injected argument.

```python
from contextlib import asynccontextmanager
from faststream import ContextRepo
from faststream.asgi import AsgiFastStream
from faststream.kafka import KafkaBroker
from faststream.middlewares import AckPolicy
from faststream_concurrent_aiokafka import (
    KafkaConcurrentProcessingMiddleware,
    initialize_concurrent_processing,
    stop_concurrent_processing,
)

broker = KafkaBroker(...)
# Register KCM on the broker before any other middleware (see DI note below)
broker.add_middleware(KafkaConcurrentProcessingMiddleware)


@asynccontextmanager
async def lifespan(_context: ContextRepo):
    await initialize_concurrent_processing(
        context=broker.context,
        concurrency_limit=20,  # max concurrent tasks (minimum: 1)
        commit_batch_size=100,  # commit after this many completed tasks
        commit_batch_timeout_sec=5.0,  # or after this many seconds
    )
    try:
        yield
    finally:
        await stop_concurrent_processing(broker.context)


app = AsgiFastStream(broker, lifespan=lifespan)


@broker.subscriber("my-topic", group_id="my-group", ack_policy=AckPolicy.MANUAL)
async def handle(msg: str) -> None: ...


# Any non-MANUAL policy (ACK_FIRST is FastStream's default) is passed through
# unchanged, not processed concurrently
@broker.subscriber("other-topic", group_id="other-group", ack_policy=AckPolicy.ACK_FIRST)
async def handle_other(msg: str) -> None: ...
```

## Core Concepts

### KafkaConcurrentProcessingMiddleware

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`.

### KafkaConcurrentHandler

The processing engine. Manages:
- An `asyncio.Semaphore` to enforce `concurrency_limit`
- In-flight task tracking via a `set[asyncio.Task]`; each task's done-callback releases the semaphore, removes the task from the set, and logs any non-cancellation exception at ERROR with a traceback
- FastStream control signals raised by a middleware registered *after* this one are absorbed before they can end the task, so they neither pin the message body via a traceback nor reach error reporters that wrap asyncio tasks. See [Limitations](#faststream-control-signals-from-a-middleware-registered-after-this-one) for which are honoured and which only log
- A `KafkaBatchCommitter` for offset commits
- An optional `ConsumerRebalanceListener` (via `handler.create_rebalance_listener()`) that flushes pending commits when partitions are revoked

This library does **not** install signal handlers — shutdown is driven by your lifespan / process manager calling `stop_concurrent_processing`.

### KafkaBatchCommitter

Runs as a background asyncio task. A streaming loop absorbs `KafkaCommitTask` objects into per-partition pending state and commits each partition's contiguous-done prefix when total pending crosses `commit_batch_size`, when `commit_batch_timeout_sec` fires, or when `commit_all`/`close` sets the flush event. Cancelled tasks are treated as a hard boundary — the offset advance stops at the cancelled task so it gets redelivered on restart (at-least-once). If the committer's task dies, `CommitterIsDeadError` is raised to callers.

## API Reference

### `initialize_concurrent_processing(context, ...)`

Create and start the concurrent processing handler; store it in FastStream's context.

| Parameter | Default | Description |
|---|---|---|
| `context` | required | FastStream `ContextRepo` instance |
| `concurrency_limit` | `10` | Max concurrent asyncio tasks (minimum: 1) |
| `commit_batch_size` | `10` | Max messages per commit batch |
| `commit_batch_timeout_sec` | `10.0` | Max seconds before flushing a batch |
| `shutdown_timeout_sec` | `20.0` | Max seconds the batch committer waits for its background task to drain before forcing cancellation |
| `max_uncommitted_tasks` | `10000` | Max tasks accepted but not yet committed before the consume path blocks (backpressure). `None` disables the bound. |

Returns the `KafkaConcurrentHandler` instance.

> **Tuning `max_uncommitted_tasks`:** each uncommitted entry holds only commit metadata — a task reference, its `TopicPartition`, offset, and consumer reference — not the message payload, so the default of `10000` is on the order of a few MB. Lower it to tighten the memory bound during a commit or broker outage, at the cost of stalling consumption sooner. Keep it `>= commit_batch_size` so size-based batching can still trigger (below that, commits fall back to the timeout/flush path); set it to `None` to disable the bound and restore unbounded buffering.

### `stop_concurrent_processing(context)`

Cancel all in-flight handler tasks, flush completed offsets via the committer, then stop the handler. Uncommitted offsets (from cancelled tasks or anything queued past a cancelled offset) are redelivered on restart — at-least-once.

### `is_kafka_handler_healthy(context)`

Returns `True` if the `KafkaConcurrentHandler` stored in `context` is running and healthy, `False` otherwise (not initialized, stopped, or committer task dead). Useful for readiness/liveness probes.

### `KafkaConcurrentProcessingMiddleware`

FastStream middleware class. Register it via `broker.add_middleware(...)`. See Quick Start for usage examples.

> **Must be outermost.** `consume_scope` fires the handler as a background task and returns `None` immediately. Any middleware that wraps it on the outside will see that premature return and misfire — wrong timing, early cleanup, or missed exceptions. Middlewares added after it (i.e. inner in the chain) run correctly inside the background task.

#### DI framework compatibility (`modern-di-faststream` and similar)

DI frameworks like `modern-di-faststream` register a broker-level middleware that creates a REQUEST-scoped dependency container around each message. If that middleware is **outer** to `KafkaConcurrentProcessingMiddleware`, its scope closes as soon as `consume_scope` returns — before the background task runs — so any dependencies resolved inside the task (database sessions, repositories, …) are created from an already-closed container. Their finalizers never run, leaving connections unreturned to the pool.

**Fix**: call `broker.add_middleware(KafkaConcurrentProcessingMiddleware)` **before** `setup_di(...)` (or any equivalent DI bootstrap call). FastStream stacks broker middlewares so the **first** registered is outermost; adding KCM first makes it wrap the DI middleware, so the DI middleware runs *inside* KCM's background task and can manage the scope lifetime correctly.

```python
broker = KafkaBroker(...)
broker.add_middleware(KafkaConcurrentProcessingMiddleware)  # registered first → outermost
modern_di_faststream.setup_di(app, container=container)  # registered after → inner to KCM
```

## How It Works

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`.

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.

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).

4. **Rebalance handling**: When Kafka revokes a partition, the `ConsumerRebalanceListener` (returned by `handler.create_rebalance_listener(flush_timeout_sec=...)`) calls `committer.commit_all()` to flush pending offsets before the partition is reassigned. The flush waits for in-flight handlers up to `flush_timeout_sec` (default 10 s) so a slow handler cannot stall the rebalance past `max.poll.interval.ms`; on timeout, the remaining in-flight messages are redelivered after reassignment (at-least-once). A future optimization may scope the wait to only the revoked partitions.

5. **Shutdown**: `stop_concurrent_processing` cancels every in-flight asyncio task, then awaits `committer.close()`. The committer treats cancelled tasks as a hard offset boundary — cancelled-and-after offsets stay uncommitted and get redelivered on restart. Total wall-clock is sub-second in normal conditions and bounded by `shutdown_timeout_sec` only as a safety net for stuck network commits.

## Limitations

### FastStream control signals from a middleware registered *after* this one

A middleware you register **after** `KafkaConcurrentProcessingMiddleware` runs
*inside* the coroutine this library dispatches as a background task, so a
FastStream control signal it raises never reaches FastStream. Every such signal
is absorbed and the message's offset is committed; what differs is whether the
library could act on it.

| raised by an inner middleware | effect | logged |
|---|---|---|
| `AckMessage` | offset commits — this *is* the ack | DEBUG |
| `RejectMessage` | offset commits — for Kafka `reject()` is `ack()` | DEBUG |
| `SkipMessage` | offset commits, processing moves on | DEBUG |
| `NackMessage` | **not honoured** — offset commits instead of being redelivered | ERROR |
| `StopConsume` | **not honoured** — the subscriber keeps consuming | ERROR |
| `StopApplication` | **not honoured** — the application keeps running | ERROR |

The three marked *not honoured* have never worked from a concurrently dispatched
handler. Before 0.6.4 they failed silently — and `StopApplication`, which
subclasses `SystemExit`, tore down the event loop outright, losing every
in-flight offset. They now log a loud ERROR naming the signal instead. If you
depend on any of them, raise it from a middleware registered **before**
`KafkaConcurrentProcessingMiddleware` (which runs outside the dispatched task),
or from outside the message-processing path entirely.

Rationale and the rejected alternatives:
[`planning/decisions/2026-07-28-control-signals-not-honoured.md`](planning/decisions/2026-07-28-control-signals-not-honoured.md).

### Calling `msg.ack()` / `msg.nack()` / `msg.reject()` directly

**These raise `RuntimeError` on the concurrent path.** Offset control belongs to
`KafkaBatchCommitter`; reaching around it silently loses data, so the middleware
refuses the call rather than letting it through.

`KafkaAckableMessage.ack()` issues a bare `consumer.commit()` with no offsets,
committing the consumer's *current fetch position* — past every in-flight task on
every assigned partition — so those messages are never processed and never
redelivered. `reject()` is an ack for Kafka and carries the same hazard under an
opposite-sounding name. `nack()` issues `consumer.seek(...)`, rewinding the
partition underneath tasks already processing it.

There is no supported way to request redelivery under concurrent processing: the
offset commits even when your handler raises. See
[`planning/decisions/2026-07-28-control-signals-not-honoured.md`](planning/decisions/2026-07-28-control-signals-not-honoured.md).

Subscribers that pass through — a `FakeConsumer` under `TestKafkaBroker`, or any
non-`MANUAL` ack policy — are unaffected, because this library is not managing
their offsets. The guards are installed only on the concurrent dispatch path, so
a passed-through subscriber never sees them.

**Still unguarded:** reaching through the message to the raw consumer, as in
`msg.consumer.commit()` or `msg.consumer.seek(...)`. The consumer is one shared
object across every message and partition, so it cannot be guarded per message.
Do not do it.

### Other

- **Batch subscribers (`batch=True`) are unsupported** — a `batch=True`
  subscriber declaring `AckPolicy.MANUAL` is rejected with an explicit
  `RuntimeError`. The concurrent path is one message → one task → one offset. A
  `batch=True` subscriber on any other ack policy simply passes through, since
  the middleware does not manage it at all.
- **`ack_policy=AckPolicy.MANUAL` is required** on subscribers you want processed
  concurrently. Every other policy passes through untouched, exactly as if the
  middleware were not registered — that is what makes a single broker-level
  `add_middleware` call safe across a mix of subscribers. `ACK_FIRST` leaves its
  offsets to aiokafka's `enable_auto_commit`; `ACK`, `REJECT_ON_ERROR` and
  `NACK_ON_ERROR` are acknowledged by FastStream's own
  `AcknowledgementMiddleware` as soon as this middleware returns. That ack is
  safe on the pass-through path — each FastStream subscriber builds its own
  `AIOKafkaConsumer`, so it touches only that subscriber's partitions and cannot
  commit past another subscriber's in-flight work, and with no background task
  "consumed" and "processed" are the same moment. It would *not* be safe if such
  a subscriber were dispatched, which is precisely why it is not.

## Migration from < 0.x

Previously, `stop_concurrent_processing` waited up to `2 × shutdown_timeout_sec` for in-flight handlers to drain to completion. The new behavior cancels them immediately. The at-least-once contract is unchanged — uncommitted offsets are redelivered on restart, the same way they always were when the handler crashed mid-task.

| What changed | Old | New |
|---|---|---|
| In-flight handler tasks on stop | drained to completion | **cancelled** |
| `KafkaConcurrentHandler.wait_for_subtasks()` | public method | removed |
| `shutdown_timeout_sec` | applied separately to handler and committer | applied to committer only |
| Signal handler installation | installed automatically | removed — own them via your lifespan / process manager |

If your handlers do non-idempotent work that's expensive to repeat, ensure your handlers are wrapped in `try/finally` so cleanup runs on `CancelledError`, or pin to the previous version of this library. To trigger shutdown on SIGTERM/SIGINT, your lifespan or main entry point must catch the signal and call `stop_concurrent_processing(broker.context)` — under uvicorn / AsgiFastStream this happens automatically through the lifespan `finally` block.

## Requirements

- Python >= 3.11
- `faststream[kafka]`

## 📦 [PyPI](https://pypi.org/project/faststream-concurrent-aiokafka)

## 📝 [License](LICENSE)

## Part of `modern-python`

Browse the full list of templates and libraries in
[`modern-python`](https://github.com/modern-python) — see the org profile for the categorized index.
