Micronaut JMS SQS acknowledgement on exception by default?

[marcinmis]

Using jakarta.jms 3.1.0 and micronaut-jms-sqs 4.1.0

Is it expected behaviour for micronaut to acknowledge the message when RuntimeException is thrown? Sample:

@JMSListener(CONNECTION_FACTORY_BEAN_NAME)
public class MessageListener {

    @Queue(value = "queue-name")
    public void r(@MessageBody MessageBody messageBody) {
        throw new RuntimeException("e");
    }
}
 

JMSListener handleMessage(Message msg) code for errors:

catch (Exception e) {
            errorHandlers.forEach(handler -> handler.handle(session, msg, e));
        }

the exception is catched by the LoggingJMSListenerErrorHandler - the default error handler - and it is swallowed:

    @Override
    public void handle(Session session, Message message, Throwable ex) {
        LOGGER.error("Failed to handle message receive: " + ex.getMessage(), ex);
    }

exception is not propagated and then in `SQSSessionCallbackScheduler:

try {
        messageListener.onMessage(message); <- **no exception thrown**
    } catch (Throwable ex) {
        LOG.warn("Exception thrown from onMessage callback for message {}",
                message.getSQSMessageId(), ex);
        callbackFailed = true;
    } finally {
        if (!callbackFailed) { <- **callback failed = false**
            if (ackMode == Session.AUTO_ACKNOWLEDGE) {
                message.acknowledge(); <- **ack done**
            }
            tryNack = false;
        }
    }

[TatyOko28]

Solution : Custom Error Handler that prevents acknowledgement (Recommended)

Create an error handler that rethrows the exception or signals failure.

Custom error handler:

import io.micronaut.jms.listener.error.JMSListenerErrorHandler;
import jakarta.inject.Singleton;
import jakarta.jms.Message;
import jakarta.jms.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Singleton
public class RethrowingJMSListenerErrorHandler implements JMSListenerErrorHandler {
    
    private static final Logger log = LoggerFactory.getLogger(RethrowingJMSListenerErrorHandler.class);
    
    @Override
    public void handle(Session session, Message message, Throwable ex) throws Throwable {
        log.error("Failed to handle message - rethrowing to prevent acknowledgement", ex);
        // Rethrow to trigger callbackFailed = true and prevent ack
        throw ex;
    }
}

Register it (application.yml):

jms:
  sqs:
    default:
      error-handler-bean-name: rethrowingJMSListenerErrorHandler

Or register globally via configuration:

import io.micronaut.context.annotation.Bean;
import io.micronaut.context.annotation.Factory;
import io.micronaut.jms.listener.error.JMSListenerErrorHandler;
import jakarta.inject.Singleton;

@Factory
public class JmsErrorHandlerFactory {
    
    @Bean
    @Singleton
    public JMSListenerErrorHandler jmsListenerErrorHandler() {
        return new RethrowingJMSListenerErrorHandler();
    }
}