Exception handling in 2.0.0 M1

[drmaciej]

Issue description

I'd like to ask about https://github.com/micronaut-projects/micronaut-jms/commit/56dc69757fa34fce958b02359b899e124f9738d5#diff-a1828ad23d4698ef382cc0a39a12e66bf58ee3ddc46021a66e7c6a97ff4b24cbR129-R130.

Particularly, we can see that now exceptions are swallowed, which means that the negative acknowledger from the underlying amazon-sqs-java-messaging-lib is not used. To me, on SQS, that's not the worst thing - the queue's visibility timeout, which I control, will do its job.

Is that the long-term plan? To swallow the exception and not propagate it further? I checked the documentation but did not find any relevant information.

[sbodvanski]

Sidenotes:

The first problem is that exception from the JMS listener is being swallowed, not propagated in the io.micronaut.jms.configuration.AbstractJMSListenerMethodProcessor.generateAndBindListener() method.

private MessageListener generateAndBindListener(Object bean,
                                                ExecutableMethod<?, ?> method,
                                                ExecutorService executor,
                                                boolean acknowledge) {

    return message -> executor.submit(() -> {
        try {
            DefaultExecutableBinder<Message> binder = new DefaultExecutableBinder<>();
            BoundExecutable boundExecutable = binder.bind(method, jmsArgumentBinderRegistry, message);
            boundExecutable.invoke(bean);
            if (acknowledge) {
                try {
                    message.acknowledge();
                } catch (JMSException e) {
                    logger.error("Failed to acknowledge receipt of message with the broker. " +
                            "This message may be falsely retried.", e);
                    throw new MessageAcknowledgementException(e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            logger.error("Failed to process a message: " + message + " " + e.getMessage(), e);
        }
    });
}

The second issue is that implementation in AbstractJMSListenerMethodProcessor does not conform to the assumption in the com.amazon.sqs.javamessaging.SQSSessionCallbackScheduler.run() method. It expects, a listener to be called in the sync way as opposed to the current implementation which submits the listener handling work to a thread executor. That is why the catch block that is responsible for triggering the negative ack mechanism is never reached.

public void run() {

....

try {
    messageListener.onMessage(message);
} catch (Throwable ex) {
    LOG.info("Exception thrown from onMessage callback for message " +
             message.getSQSMessageId(), ex);
    callbackFailed = true;
} finally {
    if (!callbackFailed) {
        if (ackMode == Session.AUTO_ACKNOWLEDGE) {
            message.acknowledge();
        }
        tryNack = false;
    }
}
...
}

There is an issue that also describes the context of this problem in a more generic way: #167.

We might want to consider deprecating the Queue/Topic executor and concurrency listener feature () that is used in this case. An alternative would be to not provide a default executor when the executor is not specified explicitly by the Queue or Topic annotations. However, it would not cover the cases when executor and concurrency options are specified.

[graemerocher]

+1 to not provide a default executor when an executor is not specified as a first step. Perhaps the executor member of the annotations can be deprecated at the same time.