Unreliable Channel caching in Sender

[robotmrv]

TC

1. Declare queue
2. Try to declare queue with different params

com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-expires' for queue 'my-queue' in vhost '/': received '10800000' but current is '86400000', class-id=50, method-id=10)
	at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:510) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:346) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:178) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:111) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:670) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQConnection.handleReadFrame(AMQConnection.java:619) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.nio.NioLoop.run(NioLoop.java:150) ~[amqp-client-5.3.0.jar:5.3.0]
	at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_102]

2. Try to declare another new queue
   ER: queue declared
   AR: got error

com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-expires' for queue 'my-queue' in vhost '/': received '10800000' but current is '86400000', class-id=50, method-id=10)
	at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:253) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQChannel.asyncRpc(AMQChannel.java:354) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQChannel.privateAsyncRpc(AMQChannel.java:315) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.AMQChannel.exnWrappingAsyncRpc(AMQChannel.java:152) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.ChannelN.asyncCompletableRpc(ChannelN.java:1565) ~[amqp-client-5.3.0.jar:5.3.0]
	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.asyncCompletableRpc(AutorecoveringChannel.java:901) ~[amqp-client-5.3.0.jar:5.3.0]
	at reactor.rabbitmq.Sender.lambda$declareQueue$8(Sender.java:190) ~[classes/:na]
	...

root cause is
https://github.com/reactor/reactor-rabbitmq/blob/34629edb1cbd532587da976c4d88062b905933c1/src/main/java/reactor/rabbitmq/Sender.java#L87

public Sender(SenderOptions options) {
...
        this.channelMono = connectionMono.map(CHANNEL_CREATION_FUNCTION).cache();
}

Sender always uses the same channel in bind(BindingSpecification), declareExchange(ExchangeSpecification), declareQueue(QueueSpecification), deleteExchange(ExchangeSpecification, boolean), deleteQueue(QueueSpecification, boolean, boolean), unbind(BindingSpecification) methods
so if it closes there is no way to use another.

[acogoluegnes]

@robotmrv Thanks for raising that up. I'm not sure there's a definitive answer here. The channel is cached to avoid using too much resource, it would be wasteful to open a channel for each operation.

I see different approaches (thinking out loud for a solution or options):

• not doing anything, and let the application deal with the error, e.g. by creating a new Sender (possibly with the same AMQP Connection) and moving on.
• transparently re-opening a channel when the cached one is closed. But how to reliably and efficiently detect the closing? And should the channel be re-opened all the time? Quite dangerous if a bug in the application is responsible for the closing, the library could open new channels repeatedly.
• handle the error but being able to replace/renew the Mono<Channel>.

WDYT @robotmrv?

[Sage-Pierce]

Chiming in: I believe I ran in to a similar issue when the AMQP cluster I was trying to produce to with a Sender was locked down and the first result of sending a message was Unauthorized with every subsequent send silently failing/dropping messages with a hidden Exception:

com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error

This seems related to keeping a cached-but-closed Connection around, but I never really dug in to figure out how to solve it in the Sender after fixing my Connection credentials.

Just FYI: I and my team make fairly extensive use of reactor-kafka and reactor-rabbitmq

[robotmrv]

@acogoluegnes , thanks for your fast reply.

I can judge just from user point of view.
Option 1 is easiest way for you, but end user would not be happy with it.

It is not clear why user needs to think about reopening the channel in one case like all declare() methods, but not for sendXXX() methods which don't cache channel.

Caching solves some performance problems (I did not do any benchmarks, so I just rely on your comment),
but it produces problems with main functionality and I am not sure that this is the right way just to ignore it.

In addition Receiver doesn't use cached channel but why it is wasteful for Sender?
As far as I understand it is because the Channel is not thread safe in some cases.
From com.rabbitmq.client.Channel

Concurrency Considerations
Channel instances must not be shared between threads. Applications should prefer using a Channel per thread instead of sharing the same Channel across multiple threads. While some operations on channels are safe to invoke concurrently, some are not and will result in incorrect frame interleaving on the wire. Sharing channels between threads will also interfere with Publisher Confirms. As such, applications need to use a Channel per thread.

But even if operation is thread safe it could influence another unrelated operation and fail it. (like in my case and @Sage-Pierce case and I think a lot other cases).

Option 2 - The main drawback - it is complicated. But if channel is cached by the lib it should be managed by the lib.
So - Cache it and deal with all related problems,
or
Do not cache by default - robust and simple solution. (e.g. it seems Spring RabbitTemplate do not cache channels or at least do not share them between threads).

In addition to "just do not cache" it could be useful expose channelCreationMono and if it is specified by user - channel management is user responsibility. (like with Schedulers).

[acogoluegnes]

@robotmrv Channel creation is a blocking operation and involves a network round-trip, so re-using channels is a must. Spring AMQP (and so RabbitTemplate) pools channels, let's say it's a variation of caching.

I think letting the user specify the Mono<Channel> makes sense too.

@simonbasle @smaldini Any thoughts on this one?

[robotmrv]

@acogoluegnes
as alternative to "always cache Channel"
maybe "Share Channel for chain of operations" could be used instead.
Something like this:

// creates channel and shares it by reactor context
Sender#shareChannel(Supplier<Mono<T>> operationsChain); 

So related operations chain (like declare Exchange - declare Queue - Bind Queue) would use the same channel.

[simonbasle]

@acogoluegnes the difficulty here is that, if I understand correctly, the ChannelAlreadyClosedException is propagated by the operation code that uses the Channel, and thus not directly visible to the channelMono itself. So there's no way for it to trigger a retry or see that it now represents an invalid cache, in a central manner.

The other challenge in the use of cache() is that there is no real way of manually invalidating it 😖

Would the AMQP client's Channel pooling be accessible and reusable here, instead of Mono#cache? How are Channel returned to the pool?

[acogoluegnes]

@simonbasle Thanks for your feeback. A proxy could wrap the Channel and re-open a new one under some circumtances (e.g. appropriate exception, etc). The question is how it would fit in Reactor; is there a way to change the cached object from cache()?

I'm not sure we can easily re-use Spring AMQP Channel pool, it's meant to be used in the Spring AMQP context and would pull a bunch of dependencies. Implementing our own pool or building it on top of some existing implementation is possible though.