fix: batch Kafka messages are a list under FastStream's test broker - #30
Merged
Conversation
…er, not just tuple _locate()'s batch-detection only checked isinstance(raw, tuple), matching real confluent-kafka's getmany() -> tuple[Message, ...] - but FastStream's own confluent test/mock broker wraps batch messages in a plain list instead, so any batch=True subscriber (og_futures, the arrow-sink CDC consumer) crashed with AttributeError: 'list' object has no attribute 'topic' the moment its retry middleware ran under test. Found while bumping a downstream service to 0.5.0 - its batch consumer test suite failed immediately. This whole code path is new in 0.5.0, so the bug never shipped before now. Widened the check to isinstance(raw, (list, tuple)). Added a parametrized regression test reproducing both shapes directly (before touching the fix, confirmed it fails on the old code with `list` and passes with `tuple`). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
Same two real shapes (list, tuple) still match, but Sequence also covers any other indexable container FastStream might switch to later without needing another patch. Verified a single confluent_kafka.Message isn't a Sequence, so the non-batch branch is unaffected. Sequence guarantees __getitem__ (unlike bare Iterable), so raw[0] stays safe - Iterable would've been too wide and introduced a real crash risk for non-indexable iterables (sets, generators). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M9p8y4kfiytRMVrzGW4rFN
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
`KafkaSubscriber`'s retry middleware (`_locate()`) only checked
isinstance(raw, tuple)to detect a batch message, matching real confluent-kafka'sgetmany() -> tuple[Message, ...]. FastStream's own confluent test/mock broker wraps batch messages in a plainlistinstead, so anybatch=Truesubscriber crashes withAttributeError: 'list' object has no attribute 'topic'the instant its retry middleware runs under test.Found while bumping a downstream service to 0.5.0 — its
batch=TrueKafka consumer tests failed immediately. This whole retry-middleware code path is new in 0.5.0, so this bug never shipped before now.Fix
Widened the isinstance check to
(list, tuple).Tests
Added a parametrized regression test that builds both shapes directly. Confirmed methodology: reverted the fix temporarily with the test in place — it fails on
list, passes ontuple(matching the bug exactly); reapplied the fix — both pass.106/106 tests pass,
pre-commit run --all-filesclean.