Requeue message on unhandled client exception

[damonsutherland]

In a MessageListener's onMessage(Message) method, if a runtime exception is thrown I was expecting the message to be requeued and redelivered. I am not seeing that happen.

Here is the setup:

• Two consumers:
  1. QoS (prefetch) is set to 1
  2. Session is AUTO_ACKNOWLEDGE and NOT transacted
  3. First consumer throws runtime exception in onMessage()
  4. Second consumer processes the message
• One Producer:
  1. Sends 10 messages

Results:

• exception is thrown
• consumer 1 fails and does not recover
• messages alternate consumers, so messages 2,4,6,8 and 10 are processed
• message 1 is delivered to consumer 1, but is never redelivered
• messages 3,5,7,9 are then sent to the alternate consumer.

In section 4.5.2 of the JMS 1.1 spec, it states that, although it is a client error to throw in onMessage, if it does happen, messages should be re-queued and redelivered. I briefly checked out the JMS Compliant Test suite you're using, and it doesn't appear this type of test is covered.

[michaelklishin]

Hi @damonsutherland.

This is how the "regular" RabbitMQ Java client used to work up to https://github.com/rabbitmq/rabbitmq-java-client/blob/master/src/main/java/com/rabbitmq/client/impl/ForgivingExceptionHandler.java.
That feature, even though it's in the spec, is counterproductive in multiple ways.

Requeueing the message is quite trivial but I'd like to suggest making this configurable, just like it is in the "regular" Java client. The default can be different ("requeue on error") if it makes more sense for JMS.

[michaelklishin]

This is a breaking change if we make "requeue on exception" the default => can go into 2.0.

[damonsutherland]

@michaelklishin, that will work perfectly! Thanks. Do you have an estimated timeframe for the 2.0 release?

[michaelklishin]

@damonsutherland it's mostly up to @acogoluegnes to decide but I'd say 4 weeks sounds doable. We haven't shipped 1.7.0 yet :)

[acogoluegnes]

@damonsutherland Could you provide some sample code that reproduces the problem?

[damonsutherland]

Sure. I have attached a test project:
rabbitmq.jms.error.zip

I am also wondering if part of the problem of not redelivering message 0 is occurring because it appears that messages, when Session.AUTO_ACKNOWLEDGE is set, are being acknowledged before being sent to the MessageListener. To prevent lost messages, should messages be acknowledged after being sent to the registered MessageListener?

    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
        logger.trace("consumerTag='{}' envelope='{}'", consumerTag, envelope);
        if (this.rejecting) {
            long dtag = envelope.getDeliveryTag();
            logger.debug("basicNack: dtag='{}'", dtag);
            this.messageConsumer.getSession().explicitNack(dtag);
            return;
        }
        /* Wrap the incoming message in a GetResponse */
        GetResponse response = new GetResponse(envelope, properties, body, 0); // last parameter is remaining message count, which we don't know.
        try {
            long dtag = envelope.getDeliveryTag();
            if (this.messageListener != null) {
                this.messageConsumer.dealWithAcknowledgements(this.autoAck, dtag);
                RMQMessage msg = RMQMessage.convertMessage(this.messageConsumer.getSession(), this.messageConsumer.getDestination(), response);
                this.messageConsumer.getSession().deliverMessage(msg, this.messageListener);
            } else {
                // We are unable to deliver the message, nack it
                logger.debug("basicNack: dtag='{}' (null MessageListener)", dtag);
                this.messageConsumer.getSession().explicitNack(dtag);
            }
        } catch (JMSException x) {
            x.printStackTrace();
            throw new IOException(x);
        } catch (InterruptedException ie) {
            ie.printStackTrace();
            throw new IOException("Interrupted while delivering message", ie);
        }
    }

see line this.messageConsumer.dealWithAcknowledgements(this.autoAck, dtag);

[michaelklishin]

Thank you, Damon! We will get to it eventually (and I really hope in no more than 2 weeks for an RC that you can try).

…

On Fri, Jun 2, 2017 at 10:22 PM, Damon Sutherland ***@***.***> wrote: Sure. I have attached a test project: rabbitmq.jms.error.zip <https://github.com/rabbitmq/rabbitmq-jms-client/files/1048725/rabbitmq.jms.error.zip> I am also wondering if part of the problem of not redelivering message 0 is occurring because it appears that messages, when Session.AUTO_ACKNOWLEDGE is set, are being acknowledged before being sent to the MessageListener. To prevent lost messages, should messages be acknowledged after being sent to the registered MessageListener? public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { logger.trace("consumerTag='{}' envelope='{}'", consumerTag, envelope); if (this.rejecting) { long dtag = envelope.getDeliveryTag(); logger.debug("basicNack: dtag='{}'", dtag); this.messageConsumer.getSession().explicitNack(dtag); return; } /* Wrap the incoming message in a GetResponse */ GetResponse response = new GetResponse(envelope, properties, body, 0); // last parameter is remaining message count, which we don't know. try { long dtag = envelope.getDeliveryTag(); if (this.messageListener != null) { this.messageConsumer.dealWithAcknowledgements(this.autoAck, dtag); RMQMessage msg = RMQMessage.convertMessage(this.messageConsumer.getSession(), this.messageConsumer.getDestination(), response); this.messageConsumer.getSession().deliverMessage(msg, this.messageListener); } else { // We are unable to deliver the message, nack it logger.debug("basicNack: dtag='{}' (null MessageListener)", dtag); this.messageConsumer.getSession().explicitNack(dtag); } } catch (JMSException x) { x.printStackTrace(); throw new IOException(x); } catch (InterruptedException ie) { ie.printStackTrace(); throw new IOException("Interrupted while delivering message", ie); } } see line this.messageConsumer.dealWithAcknowledgements(this.autoAck, dtag); — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#23 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAAEQp463cOJC4y7O80kxyMaEDZWLsRcks5sAGEKgaJpZM4Ntrjn> .

-- MK Staff Software Engineer, Pivotal/RabbitMQ

[damonsutherland]

Perfect. I did fork this project and have a fix in 1.6 that seems to address the issue. If you would like, I can submit a pull request, or just post the change (it is pretty simple). It isn't configurable though.

[acogoluegnes]

Thanks @damonsutherland for pointing this out. I'm not sure the re-delivery should be the default, the specification doesn't make much sense to me in this case (with auto-acknowledgment enabled, the onus is on the consumer to handle the message, too bad if it fails).
Nevertheless, we can add an option to comply to the specification as much as possible, this could go into 1.7.

[michaelklishin]

@damonsutherland is the solution in #30 good enough for you? That's a bit different from what we do in our "regular" Java client but it's the same fundamental idea.

[damonsutherland]

@michaelklishin Yes the solution in #30 should work. Thanks.

[acogoluegnes]

@damonsutherland you can have a try in 1.7.0.RC2.