The SimpleMessageListenerContainer does not shutdown properly

[artembilan]

We can see that from logs when we close an application normally:

2024-01-09T10:56:32.795-05:00 DEBUG 24368 --- [    container-1] o.s.a.r.listener.BlockingQueueConsumer   : Received message: (Body:'[B@756a9bb3(byte[5])' MessageProperties [headers={}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=test-supplier-queue, deliveryTag=2, consumerTag=amq.ctag-sbd2qMhyoUNhBz69Y9isVQ, consumerQueue=test-supplier-queue])
2024-01-09T10:56:32.797-05:00 DEBUG 24368 --- [    container-1] o.s.a.r.l.SimpleMessageListenerContainer : Shutting down Rabbit listener container
2024-01-09T10:56:32.806-05:00  INFO 24368 --- [    container-1] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2024-01-09T10:56:32.807-05:00 DEBUG 24368 --- [pool-3-thread-5] o.s.a.r.listener.BlockingQueueConsumer   : Received cancelOk for tag amq.ctag-sbd2qMhyoUNhBz69Y9isVQ (test-supplier-queue); Consumer@35a9e665: tags=[[amq.ctag-sbd2qMhyoUNhBz69Y9isVQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:53456/,1), conn: Proxy@2cdb5974 Shared Rabbit Connection: SimpleConnection@6d4502ca [delegate=amqp://guest@127.0.0.1:53456/, localPort=53462], acknowledgeMode=AUTO local queue size=0
2024-01-09T10:56:37.819-05:00  INFO 24368 --- [    container-1] o.s.a.r.l.SimpleMessageListenerContainer : Workers not finished.
2024-01-09T10:56:37.821-05:00  WARN 24368 --- [    container-1] o.s.a.r.l.SimpleMessageListenerContainer : Closing channel for unresponsive consumer: Consumer@35a9e665: tags=[[amq.ctag-sbd2qMhyoUNhBz69Y9isVQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:53456/,1), conn: Proxy@2cdb5974 Shared Rabbit Connection: SimpleConnection@6d4502ca [delegate=amqp://guest@127.0.0.1:53456/, localPort=53462], acknowledgeMode=AUTO local queue size=0

Pay attention to 5 sec delay between Waiting for workers to finish. and Workers not finished..
This is reported by the code in the SimpleMessageListenerContainer:

				boolean finished = this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS);
				if (finished) {
					logger.info("Successfully waited for workers to finish.");
				}
				else {
					logger.info("Workers not finished.");

So, the cancellationLock is not released from a consumer which has just reported Received cancelOk for tag.

The BlockingQueueConsumer must be fixed in its handleCancelOk() to add:

BlockingQueueConsumer.this.activeObjectCounter.release(BlockingQueueConsumer.this);

NOTE: the DirectMessageListenerContainer already has such a logic:

		private void closeChannel() {
			RabbitUtils.setPhysicalCloseRequired(getChannel(), true);
			RabbitUtils.closeChannel(getChannel());
			RabbitUtils.closeConnection(this.connection);
			DirectMessageListenerContainer.this.cancellationLock.release(this);
		}

which is called from the mentioned handleCancelOk().

[haloxinyu]

@artembilan
This issue seems to be still unresolved. By using spring-rabbit 3.15, I try to shutdown SimpleMessageListenerContainer, but still get "Workers not finished" warnning
2024-06-17T15:26:32.411Z INFO [] [SpringApplicationShutdownHook] [][][][][] [o.s.a.r.l.SimpleMessageListenerContainer.lambda$shutdownAndWaitOrCallback$2(693)] - Workers not finished.
2024-06-17T15:26:32.413Z WARN [] [SpringApplicationShutdownHook] [][][][][] [o.s.a.r.l.SimpleMessageListenerContainer.lambda$shutdownAndWaitOrCallback$1(697)] - Closing channel for unresponsive consumer: Consumer@6e18d3f9: tags=[[amq.ctag-zUuqxb9iGhy0RJvcL5ghxQ, amq.ctag-hKRvkNR0o2FzFCqw-Q06IQ]], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@10.43.32.93:5672/,1), conn: Proxy@4063af6 Shared Rabbit Connection: null, acknowledgeMode=AUTO local queue size=0

I'm trying to understand the commit 37d9641#diff-537549efc15fbf10a5941ac7f1a8648a24ffc56cccc7c46f190ccff45351b44b and I'm worried that it might not work because if a consumer is cancelled, the mainLoop() will not be executed and no chance to release itself?

			try {
				initialize();
				while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
					mainLoop();
				}
			}

[artembilan]

I think you mean like this:

	finally {
				SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer);

after that while with mainLoop().
Yes, if we don't enter to that loop, we don't release this cancelled consumer.

Please, confirm that we are on the same page, and I'll provide the fix shortly.

[haloxinyu]

@artembilan
Thank you very much for your quick reply, I think your fix should work. Please help to fix it at your convenience.

[artembilan]

BTW, @haloxinyu , we have just released version 3.1.6. Give it a shot with your project!

[haloxinyu]

@artembilan
It works good! Amazing release speed, thanks a lot!