Behaviour of forceStop when mid-processing

[izaak-birchall-landbay]

Hello,

We are seeing dual message consumption issues intermittently when a message is being processed at application shutdown.

We see the following behaviour...

1. A message is mid-way through being processed
2. The application begins graceful shutdown
3. The rabbit channel is closed in SimpleMessageListenerContainer#shutdownOrWaitAndCallback as we are using the default forceStop = false
4. Processing of the message continues within the graceful shutdown window & any transactions are successfully committed to the database.
5. ACKing the processed message fails with an error like [bbit.connection.CachingConnectionFactory]: Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 156, class-id=60, method-id=80)
6. The message is re-delivered to another instance of the service and re-processed, resulting in the database action happening twice (we have not made our database operations idempotent across the board)

I understand this is the expected behaviour when forceStop is set to false. (https://docs.spring.io/spring-amqp/reference/amqp/containerAttributes.html#forceStop) and is a part of best effort 1PC when there is a failure in publishing the message.

I tested the same flow with forceStop = true.
I expected to see that the message would be successfully processed, and would not be requeued, meaning no duplicate consumption.
I observed...

1. The PRECONDITION_FAILED log line was not produced as the channel was still open when the consumer ACKed the message
2. The ACK was successfully sent to rabbit post-processing of the message (I checked in the web UI)
3. All of the prefetched messages were requeued, including the one we ACKed
4. The ACKed message was consumed again by a different instance, leading to duplicate processing.

My confusion is around point 3. It does not make sense to me to requeue a message that we have ACKed. It would make more sense for me to requeue any messages that...

1. Did not start processing by the time application shutdown was requested
2. Were not ACKed by the time the shutdown timer has expired

Is there a way to achieve the behaviour I am looking for, where ACKed messages are not requeued with forceStop = true at application shutdown? Is this even something I would want to do? (Perhaps I have missed a downfall of this behaviour)

[artembilan]

Hi, @izaak-birchall-landbay !

Any chances that you can share with us a simple project to reproduce on our side?
I'm not why that ack'ed message is re-queued as well since the logic is like:

	/**
	 * Set to true to stop the container after the current message(s) are processed and
	 * requeue any prefetched. Useful when using exclusive or single-active consumers.
	 * @param forceStop true to stop when current message(s) are processed.
	 * @since 2.4.14
	 */
	public void setForceStop(boolean forceStop) {

where the code is like this:

		private void mainLoop() throws Exception { // NOSONAR Exception
			try {
				if (SimpleMessageListenerContainer.this.stopNow.get()) {
					this.consumer.forceCloseAndClearQueue();
					return;
				}

So, we are supposed to be done with the current message before entering that RabbitUtils.setPhysicalCloseRequired(this.channel, true); to enforce re-queuing.

Sounds like there is some race condition between ack and close, where the last one is performed on the channel faster, than the first one.

I would like to understand how you are ack'ing.
Therefore a simple project from you to play on our side would be great addition to this discussion to keep us all on the same page.

Thanks

[izaak-birchall-landbay]

Hey @artembilan,

Thanks for taking a look. This investigation has been shelved on our side so I will not be able to place any more time into researching it. I will update if some more time is allocated down the line.

Thanks