Channel management for correlated publish confirm

[arn-rio]

Hello,

We experienced some issues using correlated Rabbit confirm publishers.
Rabbit server was unstable for a while. Once restored, we were unable to publish new confirmed messages to it (the max number of channel on connection was reached and the existing channels were ignored).
The solution was to restart the service (we could force close the connection : but it only worked when channel cache size wasn't set in CachingConnectionFactory)

We investigated and found that :

• when a message is published but not confirmed, assigned channel is not used for an other message. (see spring amqp documentation )
• so CachingConnectionFactory creates new channels, and then number of channel in connection increases.
• Finally, there is no channel available on the connection : max number of channels reached
  when a new channelCacheSize and channelCheckoutTimeout are set in CachingConnectionFactory, we got 'No available channels' error, otherwise, 'The channelMax limit is reached'.

We call RabbitTemplate.getUnconfirmed() in a periodic task : the correlation data are removed, but unfortunatly, the related channel is not freed and not available for new publish.

You can observe and reproduce the issue as follows :

• a periodic task that publishes a confirmed message on Rabbit every 20 ms
• a cron job that calls RabbitTemplate.getUnconfirmed()
• on a local Rabbit server, block then unblock connection by updating memory watermark

	//block connections 
	rabbitmqctl set_vm_memory_high_watermark <low threshold>
	//unblock
	rabbitmqctl set_vm_memory_high_watermark <high threshold>

Is there a better way to manage not confirmed messages? How can we free the channels when confirmation is not received before a timeout?

Thanks

[artembilan]

Any chances that you can share a simple project to let us to reproduce the problem?
It feels like we never close those channels which are waiting for confirms, but I'd like to be on the same page with you and be able to debug the problem.
This is the logic I'm talking about:

		private void returnToCache(ChannelProxy proxy) {
			if (CachingConnectionFactory.this.active && this.publisherConfirms
					&& proxy instanceof PublisherCallbackChannel) {

				this.theConnection.channelsAwaitingAcks.put(this.target, proxy);
				((PublisherCallbackChannel) proxy)
						.setAfterAckCallback(c ->
								doReturnToCache(this.theConnection.channelsAwaitingAcks.remove(c)));
			}
			else {
				doReturnToCache(proxy);
			}
		}

I wonder if CachingConnectionFactory.resetConnection() can help you somehow...

[arn-rio]

A demo : rabbitdemo.zip

Rabbit configuration can be updated directly in source. You should reproduce the issue by blocking and unblocking the connection on the Rabbit server while this app is running.

I did some tests by calling CachingConnectionFactory.resetConnection() when the RabbitTemplate.getUnconfirmed() result is not null. This seems to work around the connection issue (need further testing on my part).

[artembilan]

OK. Created an issue: #2640.
Will provide the fix soon enough.

[arn-rio]

Thanks for your help and the reactivity.
I'll do more testing when the fix is available.

[artembilan]

You can give it a shot in your project with a SNAPSHOT available from the https://repo.spring.io/snapshot

[arn-rio]

Thanks, tested on my side and no more channel leaks.

In logs, I see that the channels are closed and the confirmations of expired messages are ignored.

Do you know when the next spring-amqp will be available and if it will be included in the next Springboot version?

[artembilan]

Sure! March 18th.
See our Release Calendar: https://calendar.spring.io.
Those Spring Boot versions scheduled for March 21th are going to include those Spring AMQP versions.

[arn-rio]

Hello,

We have deployed the Spring AMQP version with the fix in our production environment but we have now a new and different channel leak. The number of channels in the connection used by our confirmed publishers messages increased continuously while the service is running (around 100 new channels are created by day).

This chart shows the number of channels by connection from Rabbit server side. The 2 connections that are growing are the connections that manage confirm publishes.

Before upgrading Sprinq AMQP version, on Rabbit Cluster, the number of channel opened minus the number of channel closed was 0. After the update, this value has became positive.

Do you have an idea where this could come from?

[artembilan]

That's something new.
Any chances to share with us some simple project to let us to reproduce?
Thanks

[arn-rio]

We have not reproduced it for now. The context is different from the previous leak. The number of channels grows while all servers are operationnal (and not during Rabbit server incident).

[arn-rio]

The probability to reproduce the problem decreases if we increase the fixedRate value.

During the issue, publish confirmed client creates and removes many channels (the channel id in the connection is always greater)

[artembilan]

2024-05-16T13:18:18.040-04:00 DEBUG 128412 --- [ectionFactory11] o.s.a.r.c.CachingConnectionFactory       : Closing cached Channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,417)
2024-05-16T13:18:18.040-04:00 DEBUG 128412 --- [ectionFactory22] o.s.a.r.c.PublisherCallbackChannelImpl   : PendingConfirms cleared
2024-05-16T13:18:18.040-04:00 DEBUG 128412 --- [ectionFactory23] o.s.a.r.c.PublisherCallbackChannelImpl   : PendingConfirms cleared
2024-05-16T13:18:18.040-04:00 DEBUG 128412 --- [ectionFactory28] o.s.a.r.c.CachingConnectionFactory       : Released permit for 'Proxy@3f163da0 Shared Rabbit Connection: SimpleConnection@683cba3f [delegate=amqp://guest@127.0.0.1:5672/, localPort=60888]', remaining: 415

So, apparently there is unconditional permits.release() which increases the size of that Semaphore causing this leak after re-connection.

Investigating further...

[arn-rio]

It looks like what I saw.

[artembilan]

I made some fix: #2715.
It is not perfect, but at least it mitigates a leak a little bit.
I still cannot determine ho those channel may come back to the cache if we have this condition:

this.theConnection.channelListLock.lock();
				try {
					// Allow for multiple close calls...
					if (CachingConnectionFactory.this.active) {
						cacheOrClose(proxy);
					}
...

		private void cacheOrClose(ChannelProxy proxy) {
			boolean alreadyCached = this.channelList.contains(proxy);
			if (this.channelList.size() >= getChannelCacheSize() && !alreadyCached) {
				if (logger.isTraceEnabled()) {
					logger.trace("Cache limit reached: " + this.target);
				}
				try {
					physicalClose();
				}
				catch (@SuppressWarnings(UNUSED) Exception e) {
				}
			}
			else if (!alreadyCached) {
				if (logger.isTraceEnabled()) {
					logger.trace("Returning cached Channel: " + this.target);
				}
				this.channelList.addLast(proxy);
				releasePermitIfNecessary();
				setHighWaterMark();
			}
		}

Will look into that later since we have to release today at least some fix.

Thank you for your feedback!

[arn-rio]

Hello, I was on vacation last week. I've tested spring-rabbit version 3.0.14 this morning and noticed the improvement. There's still a minor leak when a new connection is established, but it disappears after a few seconds. The number of channels remains below the threshold.

Thanks for the fix.