Merge pull request #2 from modern-python/fixes

small fixes

Author: lesnik512

CLAUDE.md

@@ -15,7 +15,7 @@ just publish       # bump version to $GITHUB_REF_NAME, build, publish to PyPI
 
 Run a single test file or test by name:
 ```bash
-uv run --no-sync pytest tests/committer/test_kafka_committer.py
+uv run --no-sync pytest tests/test_kafka_committer.py
 uv run --no-sync pytest -k test_committer_logs_task_exceptions
 ```
 
@@ -43,6 +43,6 @@ Runs as a background asyncio task (spawned via `spawn()`). Collects `KafkaCommit
 
 ## Key patterns
 
-- **Singleton reset in tests**: `KafkaConcurrentHandler._initialized = False` and `._instance = None` must be reset between tests (done in conftest fixtures via `stop_concurrent_processing`).
+- **Singleton reset in tests**: `KafkaConcurrentHandler._initialized = False` and `._instance = None` must be reset between tests. Each test file does this directly in an `autouse` `reset_singleton` fixture — not via `stop_concurrent_processing`.
 - **Type suppression**: use `# ty: ignore[rule-name]` (not `# type: ignore`) for ty type checker suppressions.
 - **No `from __future__ import annotations`**: annotations are evaluated eagerly; `typing.Self`/`typing.Never` are used directly (requires Python ≥ 3.11).

README.md

@@ -29,9 +29,7 @@ app = FastStream(broker)
 async def on_startup(context: ContextRepo) -> None:
     await initialize_concurrent_processing(
         context=context,
-        concurrency_limit=20,       # max concurrent tasks (0 = unlimited)
-        commit_batch_size=100,      # commit after this many messages
-        commit_batch_timeout_sec=5, # or after this many seconds
+        concurrency_limit=20,  # max concurrent tasks (0 = unlimited)
     )
 
 
@@ -64,7 +62,16 @@ With batch commit enabled, offsets are committed per partition at the highest co
 
 ## Consumer group filtering
 
-When multiple consumer groups share a topic, messages can be tagged with a `topic_group` header. The middleware will only process messages whose `topic_group` header matches the consumer's group ID, skipping the rest.
+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.
+
+```python
+# Producer side — send to a specific consumer group only
+await broker.publish(
+    {"data": "..."},
+    topic="my-topic",
+    headers={"topic_group": "group-a"},
+)
+```
 
 ## Parameters
 

faststream_concurrent_aiokafka/middleware.py

@@ -19,7 +19,7 @@ async def consume_scope(  # ty: ignore[invalid-method-override]
     ) -> typing.Any:  # noqa: ANN401
         concurrent_processing: typing.Final[KafkaConcurrentHandler] = self.context.get(_PROCESSING_CONTEXT_KEY)
         if not concurrent_processing:
-            logger.exception("Kafka middleware. There is no concurrent processing instance in the context")
+            logger.error("Kafka middleware. There is no concurrent processing instance in the context")
             info = "No concurrent processing instance in the context"
             raise RuntimeError(info)
         if not concurrent_processing.is_running:
@@ -47,7 +47,7 @@ async def handler_wrapper() -> typing.Any:  # noqa: ANN401
 async def initialize_concurrent_processing(
     context: ContextRepo,
     commit_batch_size: int,
-    commit_batch_timeout_sec: int,
+    commit_batch_timeout_sec: float,
     concurrency_limit: int = DEFAULT_CONCURRENCY_LIMIT,
     enable_batch_commit: bool = False,
 ) -> None:

faststream_concurrent_aiokafka/processing.py

@@ -77,7 +77,7 @@ def _finish_task(self, task: asyncio.Task[typing.Any]) -> None:
             self.limiter.release()
         exc: typing.Final[BaseException | None] = task.exception()
         if exc:
-            logger.error("Kafka middleware. Kafka middleware. Task has failed with the exception", exc_info=exc)
+            logger.error("Kafka middleware. Task has failed with the exception", exc_info=exc)
         self._current_tasks.discard(task)
 
     async def handle_task(
@@ -86,28 +86,28 @@ async def handle_task(
         record: ConsumerRecord,
         kafka_message: KafkaAckableMessage,
     ) -> None:
+        if not self._is_need_to_process_message(kafka_message):
+            coroutine.close()
+            return
         if self.limiter:
             await self.limiter.acquire()
-        if self._is_need_to_process_message(kafka_message):
-            task: typing.Final = asyncio.create_task(coroutine)
-            self._current_tasks.add(task)
-            task.add_done_callback(self._finish_task)
-            if self.enable_batch_commit and self._committer:
-                try:
-                    await self._committer.send_task(
-                        batch_committer.KafkaCommitTask(
-                            asyncio_task=task,
-                            offset=record.offset,
-                            consumer=kafka_message.consumer,
-                            topic_partition=TopicPartition(topic=record.topic, partition=record.partition),
-                        )
+        task: typing.Final = asyncio.create_task(coroutine)
+        self._current_tasks.add(task)
+        task.add_done_callback(self._finish_task)
+        if self.enable_batch_commit and self._committer:
+            try:
+                await self._committer.send_task(
+                    batch_committer.KafkaCommitTask(
+                        asyncio_task=task,
+                        offset=record.offset,
+                        consumer=kafka_message.consumer,
+                        topic_partition=TopicPartition(topic=record.topic, partition=record.partition),
                     )
-                except batch_committer.CommitterIsDeadError:
-                    logger.exception("Kafka middleware. Committer is dead")
-                    await self.stop()
-                    raise
-        elif self.limiter:
-            self.limiter.release()
+                )
+            except batch_committer.CommitterIsDeadError:
+                logger.exception("Kafka middleware. Committer is dead")
+                await self.stop()
+                raise
 
     async def _check_tasks_health(self) -> None:
         to_discard: typing.Final = []
@@ -116,8 +116,6 @@ async def _check_tasks_health(self) -> None:
                 to_discard.append(task)
         for task in to_discard:
             self._current_tasks.discard(task)
-            if self.limiter:
-                self.limiter.release()
         if to_discard:
             logger.info(f"Kafka middleware. Found completed but not discarded tasks, amount: {len(to_discard)}")
 

tests/test_concurrent_processing.py

@@ -328,7 +328,9 @@ async def test_concurrent_keeps_running_tasks(handler: KafkaConcurrentHandler) -
         await task
 
 
-async def test_concurrent_releases_limiter_for_done_tasks(handler_with_limit: KafkaConcurrentHandler) -> None:
+async def test_concurrent_health_check_discards_done_tasks_without_releasing_limiter(
+    handler_with_limit: KafkaConcurrentHandler,
+) -> None:
     handler: typing.Final = handler_with_limit
 
     async def quick_task() -> str:
@@ -340,8 +342,9 @@ async def quick_task() -> str:
     assert handler.limiter
     initial_value: typing.Final = handler.limiter._value
     await handler._check_tasks_health()
+    assert task not in handler._current_tasks
     assert handler.limiter
-    assert handler.limiter._value == initial_value + 1
+    assert handler.limiter._value == initial_value  # limiter released only by _finish_task callback
 
 
 async def test_concurrent_logs_found_tasks(handler: KafkaConcurrentHandler, caplog: pytest.LogCaptureFixture) -> None: