fix: commit only partitions still assigned to the consumer

faststream_concurrent_aiokafka/batch_committer.py

@@ -86,11 +86,22 @@ def _check_is_commit_task_running(self) -> None:
     async def _call_committer(self, rc: _pending_state.ReadyCommit) -> bool:
         if not rc.offsets:
             return True
+        assigned: typing.Final = rc.consumer.assignment()
+        offsets: typing.Final = {tp: offset for tp, offset in rc.offsets.items() if tp in assigned}
+        revoked: typing.Final = [tp for tp in rc.offsets if tp not in assigned]
+        if revoked:
+            logger.warning(
+                "Skipping commit for partitions no longer assigned to this consumer, "
+                "their messages will be redelivered to the new owner: %s",
+                revoked,
+            )
+        if not offsets:
+            return False
         try:
-            await rc.consumer.commit(rc.offsets)
-        except (CommitFailedError, IllegalStateError):
+            await rc.consumer.commit(offsets)
+        except (CommitFailedError, IllegalStateError) as exc:
             # Partition no longer assigned (rebalance/revocation) — discard batch, not retryable
-            logger.exception("Cannot commit due to partition loss or rebalancing, ignoring batch")
+            logger.warning("Cannot commit due to partition loss or rebalancing, ignoring batch: %r", exc)
             return False
         except KafkaError:
             # Transient error — re-queue batch for retry on next cycle
@@ -100,7 +111,7 @@ async def _call_committer(self, rc: _pending_state.ReadyCommit) -> bool:
                 await self._messages_queue.put(task)
             return False
         else:
-            return True
+            return not revoked
 
     async def _commit_ready(self, ready_commits: list[_pending_state.ReadyCommit]) -> bool:
         # One commit per consumer, concurrently — each AIOKafkaConsumer commits its

tests/mocks.py

@@ -9,11 +9,20 @@
 from faststream_concurrent_aiokafka._pending_state import KafkaCommitTask
 
 
+class _EveryPartition:
+    def __contains__(self, _item: object) -> bool:
+        return True
+
+
 class MockAIOKafkaConsumer:
-    def __init__(self, group_id: str = "test-group") -> None:
+    def __init__(self, group_id: str = "test-group", assigned: set[TopicPartition] | None = None) -> None:
         self._group_id = group_id
+        self._assigned = assigned
         self.commit = AsyncMock()
 
+    def assignment(self) -> set[TopicPartition] | _EveryPartition:
+        return _EveryPartition() if self._assigned is None else self._assigned
+
 
 class MockAsyncioTask:
     def __init__(

tests/test_kafka_committer.py

@@ -7,7 +7,7 @@
 
 import pytest
 import pytest_asyncio
-from aiokafka.errors import CommitFailedError, KafkaError
+from aiokafka.errors import CommitFailedError, IllegalStateError, KafkaError
 from aiokafka.structs import TopicPartition
 
 from faststream_concurrent_aiokafka import _pending_state
@@ -253,6 +253,65 @@ async def test_committer_ignores_commit_failed_error(
     assert committer._messages_queue.empty()
 
 
+async def test_committer_commits_only_assigned_partitions(committer: KafkaBatchCommitter) -> None:
+    """A revoked partition in the batch must not stop the still-assigned partitions from committing."""
+    assigned_tp: typing.Final = TopicPartition(topic="t", partition=0)
+    revoked_tp: typing.Final = TopicPartition(topic="t", partition=10)
+    consumer: typing.Final = MockAIOKafkaConsumer(assigned={assigned_tp})
+    rc = _pending_state.ReadyCommit(consumer=consumer, offsets={assigned_tp: 11, revoked_tp: 21}, tasks=[])
+
+    result: typing.Final = await committer._call_committer(rc)
+
+    assert result is False
+    consumer.commit.assert_called_once_with({assigned_tp: 11})
+
+
+async def test_committer_skips_commit_when_no_partition_is_assigned(committer: KafkaBatchCommitter) -> None:
+    revoked_tp: typing.Final = TopicPartition(topic="t", partition=10)
+    consumer: typing.Final = MockAIOKafkaConsumer(assigned=set())
+    rc = _pending_state.ReadyCommit(consumer=consumer, offsets={revoked_tp: 21}, tasks=[])
+
+    result: typing.Final = await committer._call_committer(rc)
+
+    assert result is False
+    consumer.commit.assert_not_called()
+    assert committer._messages_queue.empty()
+
+
+async def test_committer_logs_revoked_partitions_as_warning(
+    committer: KafkaBatchCommitter, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Revocation is routine during rebalances, so it must not reach error reporters as an ERROR."""
+    revoked_tp: typing.Final = TopicPartition(topic="t", partition=10)
+    consumer: typing.Final = MockAIOKafkaConsumer(assigned=set())
+    rc = _pending_state.ReadyCommit(consumer=consumer, offsets={revoked_tp: 21}, tasks=[])
+
+    with caplog.at_level(logging.WARNING):
+        await committer._call_committer(rc)
+
+    records: typing.Final = [r for r in caplog.records if r.name == "faststream_concurrent_aiokafka.batch_committer"]
+    assert [r.levelno for r in records] == [logging.WARNING]
+    assert str(revoked_tp) in records[0].getMessage()
+
+
+@pytest.mark.parametrize("error", [CommitFailedError(), IllegalStateError()])
+async def test_committer_logs_rebalance_commit_errors_as_warning(
+    committer: KafkaBatchCommitter,
+    mock_consumer: MockAIOKafkaConsumer,
+    sample_task: KafkaCommitTask,
+    caplog: pytest.LogCaptureFixture,
+    error: Exception,
+) -> None:
+    mock_consumer.commit.side_effect = error
+    rc = _pending_state.ReadyCommit(consumer=mock_consumer, offsets={sample_task.topic_partition: 101}, tasks=[])
+
+    with caplog.at_level(logging.WARNING):
+        await committer._call_committer(rc)
+
+    records: typing.Final = [r for r in caplog.records if r.name == "faststream_concurrent_aiokafka.batch_committer"]
+    assert [r.levelno for r in records] == [logging.WARNING]
+
+
 # ---------- map_offsets_per_partition ----------