MessageRejectedWhileStoppingException while stopping

[nima]

Hi,
I'm using spring-amqp along with RabbitMQ message broker which receives the messages from the queue and process them. also, I need to manage the incoming traffic from the MessageConsumer which reads from the queue and pass it to the core of application, so I created a InputTrafficMonitor bean which decides whether to stop/start the incoming traffic (from the message consumers) with stop/starting the injected SimpleMessageListenerContainer bean ( with calling the corresponding stop/start methods - I don't know if it's the right way to do that or not);
so I started a stress-test and found that some messages is being lost in the case of stopping the messageListenerContainer ! I checked the logs and found bunch of MessageRejectedWhileStoppingException s as follows :

2012-10-29 14:55:19,004 WARN  [amqp.rabbit.listener.SimpleMessageListenerContainer:557] - Consumer raised exception, processing can restart if the connection factory supports it. Exception summary: org.springframework.amqp.rabbit.listener.MessageRejectedWhileStoppingException: Message listener container was stopping when a message was received

I. any clue how to handle/solve this problem ?
II. is there any better way to pause the incoming traffic from the consumers rather than stop/starting the MessageListenerContainer ?

thanks in advance

[Gary Russell]

Did you change the container's acknowledgeMode from its default (AUTO)?

When the exception is thrown the container, by default, tells the broker to requeue the message.

[nima]

Hi Gary.
nope ! the acknowledgeMode is being set to AUTO
thanks again

[Gary Russell]

Hmmm... interesting. I just ran some tests and the message isn't "lost", it's sitting in Rabbit in an un-ackd state and is not redelivered after restarting the container with container.start(). When I kill the JVM, the message goes back to 'ready' and is delivered when I restart the JVM.

I'll take a look to see what's happening; please open a JIRA ticket...

https://jira.springsource.org/browse/AMQP

[Gary Russell]

I have figured out what's going on; we have a race condition where the reject is done before the consumer is actually canceled, so messages can be delivered to the "old" (stopped) consumer and remain in an un-ackd state.

The solution is to immediately cancel the consumer when the stop() is received so when we reject any in-flight messages because the container is stopped, we don't get any unintended redeliveries.

I should be able to get the fix into the 1.1.3 release which is currently due this week.

[nima]

Hi Gary, and thanks for your reply
I'm so happy to hear that you found the problem. but, ...

so messages can be delivered to the "old" (stopped) consumer and remain in an un-ackd state

well, I ran a simple test with 300,000 incoming messages and tried to randomly stop/start the MessageListenerContainer (40 times happened), and I found that 299,519 messages has been received correcty, just 18 of them were in unacked state (based on info. in web-based management console) , and so, 463 of them were lost. I found exactly 463 MessageRejectedWhileStoppingExceptions in my log. and the (18) unacked messages were added to the final received messages after restaring my application. so, the lost messages were not in unacked state.

thanks again for your help, buddy