fix: commit only partitions still assigned to the consumer

[lesnik512]

Problem

During deploys (rolling restarts → rebalances) services log, at ERROR with a traceback:

Cannot commit due to partition loss or rebalancing, ignoring batch
IllegalStateError: Partition TopicPartition(topic='...', partition=10) is not assigned

Tasks for a revoked partition can still reach the committer after the rebalance listener's flush:

• a handler outlives flush_timeout_sec, so its task stays pending and is committed after revocation;
• the consume path holds an already-fetched message (blocked on the semaphore or max_uncommitted_tasks) and dispatches it after the flush;
• a subscriber has no rebalance listener at all.

AIOKafkaConsumer.commit validates every partition in the offsets dict and raises IllegalStateError on the first unassigned one before sending anything. Because _call_committer sends one commit per consumer, a single revoked partition discarded the offsets of every partition the consumer still owned. Those only advanced again once newer messages on the same partitions were committed, which on quiet partitions means extra redelivery on the next restart or rebalance. The ERROR-level log also turned a routine rebalance into an error-reporter event on every deploy.

Fix

• _call_committer filters the offsets by consumer.assignment() right before commit. The filter and aiokafka's own validation run in the same synchronous step before the first await, so the assignment cannot change between them. Revoked partitions are logged once at WARNING; if none remain, commit is not called.
• CommitFailedError / IllegalStateError are still caught for the races the filter cannot see (e.g. a generation change on the broker), now at WARNING without a traceback.
• _call_committer returns False whenever any partition was skipped, matching the previous "batch not fully committed" meaning.

Skipped offsets are redelivered to the partition's new owner, so at-least-once is unchanged. If the same partition is reassigned back to this consumer, committing offsets whose work already finished is still correct.

Not changed: on a transient KafkaError the whole batch is re-queued, including tasks for revoked partitions; the next round skips those with a warning.

Tests

• New unit tests: only assigned partitions are committed; no commit when nothing is assigned; revoked partitions logged at WARNING; CommitFailedError and IllegalStateError logged at WARNING. All failed before the fix.
• MockAIOKafkaConsumer gains assignment(), defaulting to every partition so existing tests keep their meaning.
• Full suite including Kafka integration (just test): 199 passed. ruff and ty clean.