fix(bunny): use consumer channel attribute instead of queue.channel#2370
Open
kivanio wants to merge 2 commits into
Open
fix(bunny): use consumer channel attribute instead of queue.channel#2370kivanio wants to merge 2 commits into
kivanio wants to merge 2 commits into
Conversation
`Bunny::Consumer#queue` may be either a `Bunny::Queue` instance or a plain
`String` (queue name) — `Bunny::Consumer#initialize` documents both as valid:
# @param [Bunny::Queue,String] queue Queue messages will be consumed from
Wrappers like Hutch register consumers via
`Bunny::Channel#basic_consume(queue_name, ...)`, which constructs a
`Bunny::Consumer` whose `queue` attribute is the queue name (a `String`).
When such a consumer received a message, the `with_process_span(queue.channel, ...)`
call in `Patches::Consumer#call` raised:
NoMethodError: undefined method 'channel' for an instance of String
@ opentelemetry-instrumentation-bunny-0.25.0/lib/opentelemetry/
instrumentation/bunny/patches/consumer.rb:14:in
'OpenTelemetry::Instrumentation::Bunny::Patches::Consumer#call'
Bunny logged the exception as "Uncaught exception from consumer" and the
delivery was neither processed nor ack'd. The consumer tag stayed registered
with the broker (so the channel looked alive), but messages piled up
indefinitely — observed in production with 30k+ errors over 6h on Hutch
consumers, requiring manual `rollout restart` to drain the backlog.
Fix: use the consumer's own `channel` attribute (also `attr_reader` on
`Bunny::Consumer`), which is the same `Bunny::Channel` instance that
`queue.channel` would have returned in the Queue-object path.
Added a regression test that subscribes via `channel.basic_consume(queue.name)`
to exercise the Hutch-style String-queue scenario end-to-end.
kivanio
requested review from
a team,
ahayworth,
arielvalentin,
dazuma,
ericmustin,
fbogsany,
hannahramadan,
kaylareopelle,
mwear,
plantfansam,
robbkidd,
robertlaurin,
simi and
xuan-cao-swi
as code owners
|
|
Comment on lines
+85
to
+100
| it 'traces consuming when consumer subscribes via queue name (String, e.g. Hutch)' do | ||
| queue = channel.queue('', exclusive: true).bind(exchange, routing_key: 'ruby.#') | ||
|
|
||
| consumer = channel.basic_consume(queue.name) { |_delivery_info, _properties, _payload| } | ||
|
|
||
| exchange.publish('San Diego update', routing_key: 'ruby.news') | ||
|
|
||
| # Wait until the publish message reached the consumer | ||
| sleep 1.0 | ||
|
|
||
| channel.basic_cancel(consumer.consumer_tag) | ||
|
|
||
| process_span = spans.find { |span| span.name == "#{topic}.ruby.news process" } | ||
| _(process_span).wont_be_nil | ||
| _(process_span.kind).must_equal(:consumer) | ||
| end |
Contributor
There was a problem hiding this comment.
Can you verify that the queue name is being correctly set as an attribute for this case.
thompson-tomo
force-pushed
the
fix/bunny-consumer-queue-as-string
branch
from
6c051e2 to
f861f73
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
3 participants
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 this changes
Patches::Consumer#callcurrently dereferencesqueue.channelto get theBunny::Channelto pass intowith_process_span. That works when theconsumer was registered through
Bunny::Queue#subscribe(wherequeueis aBunny::Queue), but fails when the consumer is registered viaBunny::Channel#basic_consume(queue_name, ...)— wherequeueis a plainString.This is exactly how Hutch and similar
Bunny wrappers subscribe, and
Bunny::Consumer#initializeitself documentsboth types as valid:
When a String-queue consumer received a message, the patch raised:
Bunny logged the exception as
Uncaught exception from consumer #<Bunny::Consumer ...>and the delivery was neither processed nor ack'd. The consumer tag stayed
registered with the broker (so the channel looked alive from the broker's
side), but messages piled up indefinitely.
In production we observed 30k+ such errors in 6h on Hutch consumers,
requiring manual rolling restarts to drain the backlog every few hours.
The fix
Bunny::Consumerexposeschannelas anattr_reader— the sameBunny::Channelinstance thatqueue.channelwould have returned in theQueue-object path. Using it directly works for both cases (queue as Queue
object and queue as String) without changing semantics for existing callers.
Test
Added a regression test in
consumer_test.rbthat subscribes viachannel.basic_consume(queue.name)(String) to exercise the Hutch-stylescenario end-to-end. The test passes after the fix and fails before it with
the
NoMethodErrorshown above.The existing test (
queue.subscribe-based path) continues to pass.Checklist
fix:)main🤖 Generated with Claude Code