GH-2653: Fix deadlock in the DirectMessageListenerContainer

[artembilan]

Fixes: #2653

When not enough channel in the cache, the DirectMessageListenerContainer.consume() returns null and adjustConsumers() goes into an infinite loop, since already active consumer does not release its channel.

• Fix DirectMessageListenerContainer.consume() to re-throw an AmqpTimeoutException which is thrown when no available channels in the cache
• Catch AmqpTimeoutException in the DirectReplyToMessageListenerContainer.getChannelHolder() and reset this.consumerCount-- to allow to try existing consumer until it is available, e.g. when this one receives a reply or times out.

[utikeev]

Is it safe to update consumerCount without a lock (or at least non-atomically)?
I mean, don't processMonitorTask and getChannelHolder execute on different threads? Even with just getChannelHolder being executed on different threads one can receive an incorrect count.

[artembilan]

@utikeev ,

thank you for the pointer!
So, changed that consumerCount to an AtomicInteger.
The point of the fix is to keep that super.setConsumersPerQueue() out of the lock.

[artembilan]

Merging this into main for today's release.
At least this fixes an immediate dead-lock problem.
Anything else we can revise later when it arises.

Thanks