diff --git a/faststream_concurrent_aiokafka/batch_committer.py b/faststream_concurrent_aiokafka/batch_committer.py index a59a5bb..6471b46 100644 --- a/faststream_concurrent_aiokafka/batch_committer.py +++ b/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 diff --git a/tests/mocks.py b/tests/mocks.py index de8a221..95ec498 100644 --- a/tests/mocks.py +++ b/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__( diff --git a/tests/test_kafka_committer.py b/tests/test_kafka_committer.py index cf5f3ba..0a61fc1 100644 --- a/tests/test_kafka_committer.py +++ b/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 ----------