4

Is there a way to gracefully stop a ListenerContainer, and its associated Consumers.

What I'm trying to achieve.

  1. Stop consuming messages.
  2. Gracefully stop ListenerContainer.
  3. Await long running consumers, and ack when finished.

I'm able to stop the ListenerContainers using consumer.stop(), but active long running consumers won't complete successfully, and processed messages won't be acked and will therefore be processed again, once the ListenerContainer has been resumed.

Output

Waiting for workers to finish.
Workers not finished.
Closing channel for unresponsive consumer: Consumer@6d229b1c

The message was processed, but not acked.

I might be able to achieve a graceful shutdown using setForceCloseChannel(false), but is it possible to verify if the cancelled consumers has finished? SimpleMessageListenerContainer.doShutDown() has a local scoped List "canceledConsumers".

user634545
  • 9,099
  • 5
  • 29
  • 40

2 Answers2

3

Increase the shutdown timeout.

See Message Listener Container Configuration.

shutdownTimeout

When a container shuts down (for example, if its enclosing ApplicationContext is closed), it waits for in-flight messages to be processed up to this limit. Defaults to five seconds.

/**
 * The time to wait for workers in milliseconds after the container is stopped. If any
 * workers are active when the shutdown signal comes they will be allowed to finish
 * processing as long as they can finish within this timeout. Defaults
 * to 5 seconds.
 * @param shutdownTimeout the shutdown timeout to set
 */
public void setShutdownTimeout(long shutdownTimeout) {

EDIT

There is now a new property forceStop to stop the container after the current message is processed, requeueing other prefetched messages. https://docs.spring.io/spring-amqp/docs/current/reference/html/#forceStop

Gary Russell
  • 166,535
  • 14
  • 146
  • 179
  • Thanks for the reply. Even with setShutdownTimeout, the last consumed message might not be acked. I believe the best option is to use setForceCloseChannel(false), but then I cannot determine if the canceled consumers has finished or not. I might have to extend the SimpleRabbitListenerContainerFactory, and re-implement/extend the SimpleMessageListenerContainer to get hold of the canceledConsumers. – user634545 May 05 '19 at 17:20
  • `>Even with setShutdownTimeout, the last consumed message might not be acked` Why do you say that? Make the timeout a large value; `stop()` will block until the last consumer thread completes. You can add an `ApplicationListener` (or an `@EventListener`method) to get an event as each consumer thread exits). See [Consumer Events](https://docs.spring.io/spring-amqp/reference/#consumer-events). You can also set the force close to false as extra protection if you want. – Gary Russell May 05 '19 at 18:09
  • No matter how high we set `setShutdownTimeout`, the consumer will continue to process newly arrived messages, until the timeout has been reached. So, if the consumer start processing a message just before the timeout has been reached, the message won't be acked. I can put together an example on github. – user634545 May 05 '19 at 19:30
  • 1
    No it will not; `doShutdown()` cancels the consumer so no more messages will arrive for it. Furthermore, the consumer runs in a loop `while(isActive()...` and `shutDown()` sets `active` to false before calling `doShutDown()`. – Gary Russell May 05 '19 at 20:01
  • You are correct, my mistake. The "issue" might only occur if the `setShutdownTimeout` is less than the initial `numberOfMessages` * `consumerExecutionTimePerMessage`. I'm fine with that. Thanks. – user634545 May 05 '19 at 20:18
  • 1
    There is now a new property `forceStop` to stop the container after the current message is processed, requeueing other prefetched messages. https://docs.spring.io/spring-amqp/docs/current/reference/html/#forceStop – Gary Russell Jul 27 '23 at 17:09
1

To complete @user634545's last comment, we have to deal with 2 properties :

These parameters must be ajusted in order to validate the following :

prefetchCount < (shutdownTimeout / consumerExecutionTimePerMessage)

It would mean that after receiving the shutdown order, the consumer should be able to consume and aknowledge every prefetched messages.

Alexandre D.
  • 11
  • 3
  • Shouldn't the formula be `prefetchCount < (shutdownTimeout / consumerExecutionTimePerMessage)`? – Nischo Mar 01 '22 at 12:18
  • Absolutely, you're rigtht, I edit the formula. – Alexandre D. Mar 02 '22 at 13:08
  • There is now a new property `forceStop` to stop the container after the current message is processed, requeueing other prefetched messages. https://docs.spring.io/spring-amqp/docs/current/reference/html/#forceStop – Gary Russell Jul 27 '23 at 17:10