ImmediateAcknowledgeAmqpException keeps the message in the queue

[alexey-anufriev]

In what version(s) of Spring AMQP are you seeing this issue?

3.1.1

Describe the bug

When ImmediateAcknowledgeAmqpException is thrown from the code of the listener
it is expected that the message that has just been consumed by the listener
will be automatically acknowledged and removed from the queue.

To Reproduce

Clone https://github.com/alexey-anufriev/spring-amqp-immediate-ack
Run ./mvnw clean test

Expected behavior

Queue is empty

Sample

https://github.com/alexey-anufriev/spring-amqp-immediate-ack

[artembilan]

Apparently that doesn't work for container.setAcknowledgeMode(AcknowledgeMode.MANUAL);:

boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();

So, instead of throwing that exception from your @RabbitListener, you have to do channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);.
Technically those are exactly the same, but since you are in a manual mode, you cannot rely on auto-ack: just doesn't match according to the logic.

Closed as Works as Designed

[alexey-anufriev]

This scenario I shared is a simple version of the problem, but what if I want to immediate-ack from the Error Handler? For example, I have a custom Error Handler that has to map business exceptions into AQMP-ones. I have already successfully used AmqpRejectAndDontRequeueException but now want to use ImmediateAcknowledgeAmqpException there as well.

What should I do in this case?

[artembilan]

The MANUAL ack mode means that you handle errors in your listener method.
The error handler on the listener is not aware of the Channel.

[alexey-anufriev]

Thank you @artembilan for the explanation!

[alexey-anufriev]

@artembilan, one last question: how can I handle errors that happen in the post-processor but still within the listener method?

[artembilan]

Yeah...
I see your point:

			if (this.afterReceivePostProcessors != null) {
				for (MessagePostProcessor processor : this.afterReceivePostProcessors) {
					message = processor.postProcessMessage(message);
					if (message == null) {
						throw new ImmediateAcknowledgeAmqpException(
								"Message Post Processor returned 'null', discarding message");
					}
				}
			}

This way it does not reach listener method yet.
We probably need to relax the mention logic in the commitIfNecessary() and call this.channel.basicAck(deliveryTag, true); independently of the ack mode...
Or make a soft change to call this ack here instead of deferring to the exception flow control.

Reopening as this sounds like valid request.

Thanks

[alexey-anufriev]

@artembilan, thanks again for your help.

[artembilan]

So, the fix is in.
Now SimpleMessageListenerContainer has a new option enforceImmediateAckForManual you can set to true - and those ImmediateAcknowledgeAmqpException will trigger and ack on the message even if AcknowledgeMode.MANUAL.
Feel free to pull 3.1.2-SNAPSHOT for testing on your side.

Thanks

[alexey-anufriev]

@artembilan, thank you for your effort! I will take a look at the snapshot version. I hope this will be included in the next Spring Boot release (we usually bring it all together).

[artembilan]

Correct. We have it scheduled for February, 19th.