manual message acknowledgement is broken in 2.x with a MessageListener and requeueOnMessageListenerException=true

[lepurvis-fadv]

Describe the bug

Calling Message.acknowledge() from inside MessageListener.onMessage() is basically a no-op if RMQConnectionFactory.setRequeueOnMessageListenerException(true) and the RMQSession is created with CLIENT_INDIVIDUAL_ACKNOWLEDGE or CLIENT_ACKNOWLEDGE - the acknowledgement is never sent to the rabbitmq server.

In com.rabbitmq.jms.client.RMQSession.acknowledge(long) line 1336 unackedMessageTags is always empty.

This is due to com.rabbitmq.jms.client.MessageListenerConsumer.handleDelivery(String, Envelope, BasicProperties, byte[]) not calling dealWithAcknowledgments() before calling the MessageListener when this.requeueOnMessageListenerException is true. dealWithAcknowledgments() in turn calls back to the session, which is what populates unackedMessageTags.

The fix is more involved than just reordering dealWithAcknowledgments() as is done when this.requeueOnMessageListenerException is false, as when it is true, it shouldn't be called if the message needs to be nacked.

So somehow com.rabbitmq.jms.client.RMQSession.unackedMessageTags needs to be populated before dispatching a message to a MessageListener.

Reproduction steps

1. Configure RMQConnectionFactory with setRequeueOnMessageListenerException(true)
2. Create a session with RMQSession.CLIENT_INDIVIDUAL_ACKNOWLEDGE
3. Create a MessageConsumer on a Queue
4. Set a MessageListener on the MessageConsumer
5. Observe that Message.acknowledge() from onMessage() is a silent no-op (no exception is thrown)

Expected behavior

javax.jms.Message.acknowledge() should acknowledge the message as defined by the JMS spec, i.e. the rabbitmq server should receive an ack for the message.

Additional context

To reproduce, unzip the attached rmq-jms-issue.zip and cd into the rmq-jms-issue directory. Then:

./gradlew run

Note that only one message is delivered to the MessageListener and no outbound ack for the message is logged by the TrafficListener. Then:

./gradlew run -DrequeueOnMessageListenerException=false

Note that now all messages are delivered and outbound acks are logged.

This small test application expects a rabbitmq server on localhost on the default port with username/password both "guest". It uses "test" for the queue, exchange, and routing key, but any queue with text messages will work with the example code as-is.

rmq-jms-issue.zip

Originally posted by @lepurvis-fadv in #582

[michaelklishin]

Thank you for putting together an executable example!