feat: pause Kafka partitions instead of blocking backoff for non-keyed messages#31
Open
aradng wants to merge 3 commits into
Open
feat: pause Kafka partitions instead of blocking backoff for non-keyed messages#31aradng wants to merge 3 commits into
aradng wants to merge 3 commits into
Conversation
…d messages Non-keyed topics have no ordering requirement, so blocking a whole consumer's poll loop while one message backs off needlessly delays unrelated messages on other partitions. For a message with no Kafka key, KafkaSubscriber now pauses just that partition (confluent_kafka's Consumer.pause/.resume) and keeps polling, instead of sleeping inline. Keyed messages are unchanged. See docs/adr/01-kafka-non-keyed-retry-backoff.md for the alternatives considered (numbered retry topics, an external delay store, max_workers concurrency) and why each was rejected. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
- pause() failure no longer swallows the original handler exception - falls back to inline backoff instead, preserving the NACK-and-redeliver contract. - resume() retries up to 3 times with backoff before giving up, instead of silently stranding a partition paused forever on the first failure. - KafkaSubscriber now cancels pending resume tasks when the broker stops, instead of leaving them to fire against an already-closed consumer. - _is_keyed now checks every record in a batch (any() defaults to the safe/blocking choice) instead of classifying a mixed batch off record 0. - collapsed _pause_partition/_resume_partition's duplicated TopicPartition-build-and-run_in_executor shape into one _set_partition_paused helper, reusing FastStream's own run_in_executor instead of hand-rolling it. - tests: fixed-consumer thread pools now get shut down, added coverage for both new failure paths, and the integration test now asserts partition 0's original message actually gets redelivered and succeeds after resume (previously only checked partition 1 wasn't blocked), plus polls for topic readiness instead of a fixed sleep. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
- moved keyed/non-keyed pause-resume test coverage from SimpleNamespace fakes into real-broker testcontainer tests (test_pause_resume_backoff.py) - a thin proxy wraps the real confluent Consumer to record/force-fail pause()/resume(), everything else (broker, thread pool, message flow) stays real. Each test uses its own topic to avoid cross-test message replay from auto_offset_reset="earliest" on a shared topic. - test_backoff.py reverts to its original fake-message shape - it only ever exercises the keyed (inline-sleep) branch now, so none of the consumer/thread-pool machinery is needed there anymore. - trimmed the new inline comments in depends.py to one-line pointers at docs/adr/01-kafka-retry-backoff.md instead of re-explaining the reasoning at each call site. - renamed the ADR to docs/adr/01-kafka-retry-backoff.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
KafkaSubscribernow pauses just the failing(topic, partition)viaconfluent_kafka.Consumer.pause()/.resume()and keeps polling, instead ofasyncio.sleep-ing inline. Keyed messages are unchanged (still block inline - their partition is already one ordered stream, so blocking it costs nothing extra).docs/adr/01-kafka-non-keyed-retry-backoff.mdrecords the four options considered (numbered retry topics, an external delay store,max_workersconcurrency, partition pause/resume) and why the first three were rejected.docs/signals.md's Kafka retry section is trimmed to the operational facts, pointing to the ADR for the "why".Test plan
tests/kafka/test_backoff.py: extended with keyed-vs-non-keyed branch coverage (pause/resume calls recorded on a fake consumer, repeated-failure backoff still doubles correctly).tests/kafka/test_pause_resume_backoff.py: new real-broker (testcontainers) integration test - publishes to partition 0, waits for it to actually fail, then publishes to partition 1 and asserts it's processed well under the backoff delay. Verified this test fails against the pre-fix code (partition 1 blocked ~4.7s) and passes against the fix.poetry run pre-commit run --all-files- all hooks pass.poetry run pytest -n auto- 110 passed.🤖 Generated with Claude Code