Spring AMQP Consumer pauses after running for sometime [AMQP-621]

[spring-operator]

Mithra Kottravel opened AMQP-621 and commented

We have a 2 node RabbitMQ cluster with Ha-all policy. We use Spring AMQP in our application to talk to RabbitMQ. Producer part is working fine, but consumer works for some time and pauses. Producer and consumer are running as different applications. More information on Consumer part.

• we use SimpleMessageListenerContainer with ChannelAwareMessageListener, use Manual ack mode and default prefetch(1)
• In our application we create queue (on-demand) and add it to the listener
• When we started with 10 ConcurrentConsumers and 20 MaxConcurrentConsumers, consumption happens for around 15 hours and pauses. This situation happens within 1 hour when we increase the MaxConcurrentConsumers to 75.

On RabbitMQ UI, we see channels with 3/4 unacked messages on the channel tab when this situation occurs, until then it just have 1 unacked message.

Having heartbeat set to 60 did not help improve this situation.

We dynamically add and remove queues to SimpleMessageListenerContainer and we suspect this is causing some problem, because every time we add or remove a queue from the listener, all the BlockingQueueConsumer are removed and created again.
@GaryRussell, suggested to create Jira ticket. Please let me know if you need more information.

Spring AMQP version: 1.6.0.RELEASE
Thread Dump: pastebin.com/UrBLfn2C
Stackoverflow Link: http://stackoverflow.com/questions/38134687/spring-amqp-consumer-pauses-after-running-for-sometime?noredirect=1#comment63848781_38134687

---

Affects: 1.6 GA

Reference URL: http://stackoverflow.com/questions/38134687/spring-amqp-consumer-pauses-after-running-for-sometime

Referenced from: pull request #432

[spring-operator]

Artem Bilan commented

Looks like we have a race condition when we initiate:

consumer.getKey().basicCancel();
consumer.setValue(false);

and than in the AsyncMessageProcessingConsumer loop we have a false for both flags:

while (isActive(this.consumer) || this.consumer.hasDelivery()) {

And just after that we receive a new delivery in the BlockingQueueConsumer.InternalConsumer.handleDelivery(). Two new deliveries... And in this case we are blocked on the

BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body));

Just an info to share.

Need more time for investigation and thinking...

[spring-operator]

Mithra Kottravel commented

@Artem Bilan / @Gary Russell
Simple Clarification: when we do channel.basicCancel(), I know this will prevent RabbitMQ server from sending new messages to the client, but does it allow acks to be sent from client to server after basicCancel has completed?

We wanted to used RabbitMq in our Production, so can you suggest any workaround until this issue is fixed?

[spring-operator]

Artem Bilan commented

Mithra Kottravel ,sorry for delay.

Well, actually that should work as you expected.
The Consumer isn't related to the Delivery and they really are handled separately.
And exactly that was our intention when we had changed logic from Thread blocking to just proceed after basicCancel().
Now in-flight messages are pushed to the listener even if we have just cancelled Consumer:

while (isActive(this.consumer) || this.consumer.hasDelivery()) {
     try {
            boolean receivedOk = receiveAndExecute(this.consumer); // At least one message received

But you should call channel.basicAck() exactly in the listener thread, making it blocked. Otherwise, according the cancelled state, we go out from the listener loop and go to the this.consumer.stop();, which closes the Channel as well. Yes, the last one may just be returned to the cache and remains open, but we shouldn't rely on this impermanent behavior, especially when we simply exceed the cache limit with high concurrency.

I'll try to come up with some fix for the issue today, by the way...

[spring-operator]

Mithra Kottravel commented

@Artem Bilan, Thanks for the quick fix. Do you know when it will be released to Maven?

[spring-operator]

Artem Bilan commented

The next Monday!

Although we'd be glad to see if you are able to test your application with the current 1.6.1-BUILD-SNAPSHOT from the:

<repository>
	<id>spring-snapshots</id>
	<name>Spring Snapshots</name>
	<url>http://repo.spring.io/libs-snapshot-local</url>
	<snapshots>
		<enabled>true</enabled>
	</snapshots>
	<releases>
		<enabled>false</enabled>
	</releases>
</repository>

That would give us one more chance to fix anything else if that.

Thank you for feedback anyway!