refactor: split signals into rabbit/kafka subpackages - #29
Merged
Conversation
…ubscriber Kafka consumers using AckPolicy.NACK_ON_ERROR had zero backoff on exception -- a deterministic (poison) message failure gets redelivered essentially as fast as the consumer can poll, with no delay, no cap, no dead-letter. RabbitSubscriber already handles this correctly (DLX queue chain, exponential backoff, 5s*2^n capped at 24h) -- Kafka's consumer never got the same treatment. Kafka has no per-message TTL primitive to build a delay-queue chain from the way Rabbit does, so this takes a simpler route: an ExceptionMiddleware handler (same pattern as RabbitSubscriber's _exc_handler) that sleeps `base_delay * 2 ** (attempt - 1)` seconds (capped at max_delay) before re-raising. NACK_ON_ERROR's redelivery is throttled directly instead of spinning at max poll rate; the exception still propagates for Sentry/OTel visibility either way. Retry state is tracked per (topic, partition), not per-offset: Kafka delivers one partition's messages strictly in order, so a partition can only ever be stuck retrying one offset at a time. Tracking coarser than that keeps the in-memory state bounded by the consumer's own partition assignment rather than growing with every distinct message that's ever failed. KafkaSubscriber's constructor gains the same base_delay/max_delay/ exceptions parameters RabbitSubscriber already has, all defaulted, so this is backwards compatible with every existing single-arg call site (grepped this repo -- fastloom/launcher/utils.py and tests/kafka/conftest.py both just do KafkaSubscriber(settings)). Found while investigating a production email-spam incident in a downstream service: a deterministic bug (unrelated to this fix) turned into hundreds of retries per hour purely because nothing throttled the redelivery. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015VFwpmqkP4XhgJzbdk1vKh
Strip explanatory docstrings/comments (params-only, matching RabbitSubscriber's style); register the retry middleware directly via KafkaRouter's middlewares= constructor arg instead of router.broker. add_middleware; clear _retry_state per (topic, partition) on a successful consume instead of only ever growing it, closing the unbounded-memory concern. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
- guard against ACK_FIRST (offset commits before handler runs, so backoff would silently no-op) with a loud RuntimeError instead - drop bare asserts on broker-sourced topic/partition/offset, matching FastStream's own nack() tolerance for None - don't clobber a different in-flight offset's retry state on success (matters once max_workers>1) - full-jitter both Rabbit and Kafka backoff delays to avoid a retry hail storm; Rabbit jitters via per-message AMQP expiration so DLX queue naming/TTL (and reuse) stay keyed on the plain delay - lower Kafka max_delay default 24h -> 30min: this sleep blocks the whole subscriber's poll loop, not just the failing partition Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
Full jitter (0 to delay) didn't respect the exponential curve - could collapse a high-attempt delay near zero. Switched to +/-10% around the capped exponential delay, and factored the shared math (backoff_delay, with_jitter) into fastloom/signals/utils.py so Rabbit and Kafka can't drift apart on the formula. Rabbit's DLX queue naming/TTL still key off the plain (unjittered) delay for reuse; only the per-message AMQP expiration gets jittered, so a positive jitter draw is silently clamped back to `delay` by the queue's own TTL (RabbitMQ takes whichever is lower) - only the negative half of the range is actually observable there. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
One function instead of two composed calls - jitter defaults on for the actual sleep/expiration, Rabbit's queue-naming call opts out (jitter=False) since it needs the plain delay for stable bucketing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
…o signals.rabbit (0.5.0)
BREAKING: fastloom.signals.{depends,settings,middlewares,healthcheck}
moved to fastloom.signals.rabbit.* to mirror signals.kafka.* - every
import fixed across fastloom, tests, docs, README, and the
add-rabbit-subscriber skill.
KafkaSubscriber now:
- defaults the broker-wide ack_policy to NACK_ON_ERROR (reaching into
router.broker.config.broker_config.ack_policy - the one mutable
field the read-only composed ack_policy property actually reads
from, since KafkaRouter's own constructor doesn't expose it).
Individual @subscriber(...) calls still override; one that
deliberately stays on ACK_FIRST just gets its backoff silently
skipped, no more masking the real exception with a RuntimeError.
- exposes allow_auto_create_topics (default True, wire TC.general.DEBUG
from the launcher to disable outside dev) and acks/enable_idempotence
(default 1/False) as real constructor params instead of only being
reachable via raw KafkaBroker.
Added tests/test_signals_utils.py to directly verify
exponential_backoff's doubling and cap behavior across several
attempts, not just one.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
aradng
changed the title
aradng
changed the title
…ests, add Kafka subscriber skill - fastloom.signals.utils -> fastloom.utils: exponential_backoff has zero pubsub coupling, belongs with date.py/crypto.py/types.py, not nested under signals - tests/test_rabbit_backoff.py -> tests/rabbit/test_backoff.py, mirroring tests/kafka/ - add plugins/fastloom-sdk/skills/add-kafka-subscriber, adapted from add-rabbit-subscriber for Kafka's actual API (no subscriber() wrapper, ack_policy/auto_offset_reset/max_workers gotchas); update plugin.json description and fastloom-reference's cross-reference lists Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
…rkers advice, idempotence/acks conflict Verified against the actual FastStream/librdkafka source and a live confluent_kafka.Producer construction, not just review-agent claims: - max_delay default 1800 -> 240: FastStream's own max_poll_interval_ms default is 5min, and the backoff sleep never calls poll() in between, so anything near the old 30min cap got rebalanced out of the consumer group mid-backoff - the opposite of throttling. Doc explains the coupling and how to raise both together if you want longer backoff. - pulled the max_workers>1 mitigation advice from the docstring, docs, and the new skill: KafkaMessage.ack() commits the consumer's current position, not a specific offset, so a concurrent handler acking a later offset can commit past an earlier one still asleep in backoff - a crash in that window permanently skips it. - enable_idempotence=True now forces acks="all" instead of trusting the caller to also pass acks="all" themselves - confirmed empirically that librdkafka's Producer rejects enable.idempotence with any other acks value at construction time (acks=1 raises KafkaException immediately). - get_kafka_router's allow_auto_create_topics/acks/enable_idempotence are now keyword-only and required (no restated defaults vs. KafkaSubscriber.__init__, the single source of truth for those). - _retry_state -> NamedTuple, pop(key, None) -> del, dropped an unnecessary AckPolicy alias, hoisted a repeated header read in rabbit/depends.py, fixed a stale signals.healthcheck path in docs/healthcheck.md the earlier rename swept past, test helper now looks up _RetryMiddleware by name instead of a hardcoded index. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
…(review-2) - _locate() returns one flat _MessageKey(topic, partition, offset) NamedTuple instead of a nested (key, offset) tuple - _backoff/ _clear_retry_state each take one argument instead of splatting two - _retry_state is now dict[_Partition, _RetryState], both NamedTuples, instead of a bare tuple[str, int] key - consume_scope early-returns call_next(msg) directly when topic/ partition/offset are missing, instead of wrapping that case in the same try/except and special-casing it in the except branch. Drops the "missing metadata" log warning as a result - no seam left to log from without reintroducing the try/except this was meant to avoid. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
aradng
force-pushed
the
fix/kafka-consumer-retry-backoff
branch
from
July 21, 2026 13:45
5e7cb4b to
5847a12
Compare
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.
What
Kafka consumers using
AckPolicy.NACK_ON_ERRORhad zero backoff on exception — a deterministic (poison) message failure gets redelivered essentially as fast as the consumer can poll, with no delay, no cap, no dead-letter.RabbitSubscriberalready handles this correctly (DLX queue chain, exponential backoff,5s*2^ncapped at 24h) — Kafka's consumer never got the same treatment.Why not a DLX chain like Rabbit
Kafka has no per-message TTL primitive to build a delay-queue chain from the way Rabbit does (no
x-message-ttl-equivalent). This takes the simpler, Kafka-appropriate route instead: anExceptionMiddlewarehandler (same pattern asRabbitSubscriber._exc_handler) that sleepsbase_delay * 2 ** (attempt - 1)seconds (capped atmax_delay) before re-raising.NACK_ON_ERROR's redelivery is throttled directly instead of spinning at the consumer's max poll rate — same outcome (throttled retries), simpler mechanism (no new topics/infra), and the exception still propagates for Sentry/OTel visibility either way, matching Rabbit's own choice to always re-raise.Retry state design
Tracked per
(topic, partition), not per-offset: Kafka delivers one partition's messages strictly in order, so a partition can only ever be stuck retrying one offset at a time. Tracking coarser than per-offset keeps the in-memory state bounded by the consumer's own partition assignment, rather than growing with every distinct message that's ever failed over the process's lifetime.Compatibility
KafkaSubscriber's constructor gains the samebase_delay/max_delay/exceptionsparametersRabbitSubscriberalready has, all defaulted — backwards compatible with every existing single-arg call site (grepped this repo:fastloom/launcher/utils.pyandtests/kafka/conftest.pyboth just doKafkaSubscriber(settings)).Tests
tests/kafka/test_backoff.py(new, pure unit — no broker needed): delay doubles on repeated failures of the same offset, capped atmax_delay; resets to attempt 1 on a new offset; tracks partitions independently.tests/kafka/suite: 23 passed (existing tests unaffected).mypy/ruffclean on both changed files. (Note: a handful of pre-existing mypy errors in this same file, in the unrelated tombstone-patching code above my changes, are untouched by this PR.)Context
Found while investigating a production email-spam incident in a downstream service: a deterministic bug (unrelated to this fix, already fixed there) turned into hundreds of retries per hour purely because nothing throttled the redelivery. That's a systemic gap in this library, not just that one service's bug.