0

У меня возникла проблема с тем, что не работает ExceptionListener при ошибке Rabbit JMS.

Имеется вот такая настройка:

...

private DefaultJmsListenerContainerFactory initContainerFactory(ConnectionFactory factory) {
    final DefaultJmsListenerContainerFactory containerFactory = new CustomJmsListenerContainerFactory(properties.getJms());
    containerFactory.setRecoveryInterval(DEFAULT_RECOVERY_INTERVAL);
    containerFactory.setExceptionListener(queueLogService::addLogByException);
    if (StringUtils.isNotEmpty(properties.getJms().getConcurrentListenersSize())) {
        containerFactory.setConcurrency(properties.getJms().getConcurrentListenersSize());
    }

    configurer.configure(containerFactory, factory);
    return containerFactory;
}

...

Метод addLogByException():

@Override
public void addLog(QueueRequest request) {
    if (logs.size() + 1 >= maxLogs) {
        logs.poll();
    }

    logs.add(request);
}

@Override
public void addLogByException(Throwable exception) {
    addLog(new QueueRequest(exception.getLocalizedMessage(), true));
}

Затем я намеренно создаю ошибку следующим путём:

Так как у меня стоит протокол JMS, то я помещаю в очередь сообщение в неверном формате (AMQP) и Rabbit JMS начинает логгировать очень-очень много исключений с ошибкой invalid stream header: 67646366:

2022-06-07 10:42:41.071  WARN 16352 --- [ult_wm-queue-25] o.s.j.l.DefaultMessageListenerContainer  : Setup of JMS message listener invoker failed for destination 'wm-queue' - trying to recover. Cause: invalid stream header: 67646366
2022-06-07 10:42:41.084  INFO 16352 --- [ult_wm-queue-26] o.s.j.c.CachingConnectionFactory         : Encountered a JMSException - resetting the underlying JMS Connection

com.rabbitmq.jms.util.RMQJMSException: invalid stream header: 67646366
    at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1140) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.convertJmsMessage(RMQMessage.java:913) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.convertMessage(RMQMessage.java:907) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:356) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:269) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at org.springframework.jms.support.destination.JmsDestinationAccessor.receiveFromConsumer(JmsDestinationAccessor.java:132) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveMessage(AbstractPollingMessageListenerContainer.java:418) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:303) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:257) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1237) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1227) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1120) ~[spring-jms-5.3.13.jar:5.3.13]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: java.io.StreamCorruptedException: invalid stream header: 67646366
    at java.base/java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:946) ~[na:na]
    at java.base/java.io.ObjectInputStream.<init>(ObjectInputStream.java:385) ~[na:na]
    at com.rabbitmq.jms.util.WhiteListObjectInputStream.<init>(WhiteListObjectInputStream.java:90) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1114) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    ... 12 common frames omitted

ВОПРОС: метод addLogByException() не вызывается, хотя судя по названию обработчик исключений, наверное, должен. Почему?

1 ответ 1

0

Проблему удалось решить.

Оказалось, что org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer#configure перетирает ExceptionListener в месте org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer.java:117.

В преведенном выше коде, я сначала устанавливал слушателя, а потом вызывал метод configure(), поэтому у меня ничего не работало :(

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.