[BEAM-7434] [BEAM-5895] and [BEAM-5894] Upgrade to rabbit amqp-client 5.x #9900
Conversation
Note: switches to a purely pull-based api. See https://www.rabbitmq.com/api-guide.html#getting
| @@ -376,11 +372,6 @@ public void finalizeCheckpoint() throws IOException { | |||
| this.source = source; | |||
| this.current = null; | |||
| this.checkpointMark = checkpointMark != null ? checkpointMark : new RabbitMQCheckpointMark(); | |||
| try { | |||
There was a problem hiding this comment.
I've addressed this in another PR https://github.com/apache/beam/pull/9820/files#r335994291 but tl;dr: the ConnectionHandler used in the constructor and close is not the same one used in start. I now defer initializing the class-level ConnectionHandler to start as it feels more in-line with start/stop semantics.
| QueueingConsumer.Delivery delivery = consumer.nextDelivery(1000); | ||
| Channel channel = connectionHandler.getChannel(); | ||
| // we consume message without autoAck (we want to do the ack ourselves) | ||
| GetResponse delivery = channel.basicGet(queueName, false); |
There was a problem hiding this comment.
Important change: previously the caller would wait for up to 1s for a message to be available. This PR's implementation would return false immediately (technically after a network hop). The semantics of this change is unclear to me but from what I can tell, this should be "fine" at worst and an improvement at best.
There was a problem hiding this comment.
@chamikaramj @dpmills @reuvenlax might have good guidance here
Primarily addresses BEAM-7434 which notes that the use of the java RabbitMQ client's
QueueingConsumeris deprecated in the 4.x line and is slated for removal in 5.x.This PR:
Channel.basicGetdirectly (pull api)Notes:
The 4.x use of QueueingConsumer was (presumably) done for two reasons: 1) thread safety between Channel/Consumer (see history notes in QueueingConsumer); 2) to apply backpressure on
pushby pushing to a blocking queue. Number 1 is no longer an issue, and I've circumvented number 2 by using a pull-based api rather than the consumer push-based.There are some tradeoffs in place in this PR. Previously, using the push-based, Consumer api, messages could be internally buffered/queued by each shard of the unbounded source, which saves round-trip latency if there are messages ready. Both before and after my changes, "prefetch" values were set at the default "unlimited" (0) value so messages would enqueue/buffer as soon as they're able. The trade-off there would be higher memory usage, although it's unlikely this is a practical concern for many.
Using the pull-based api (
channel.basicGet), each message pull will require a round-trip to the rabbitmq server, albeit on an already-open Connection. It should have a nice property where pulling is 'fairer' than the previous approach, as each shard will only request a message when explicitly prompted to by the Runner rather than potentially being unacked/buffered within some other shard.I'm expecting this API should not cause any real-world problems and the throughput should be high enough without local buffering. If local buffering/minimal round-trip time is of concern, I would suggest a follow-up PR allowing for configuring prefetch (and perhaps other QoS settings).
Note this is not binary compatible with previous releases, both in terms of the underlying RabbitMQ API as well as the wire format of the Beam RabbitMqIO implementation as it switches from
QueueingConsumer.DeliverytoGetResult.R: @jbonofre
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.See the Contributor Guide for more tips on how to make review process smoother.
Post-Commit Tests Status (on master branch)
Pre-Commit Tests Status (on master branch)
See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.