Spring boot rabbitMq setExceptionHandler

0
votes

How can I convert the following code using the Spring framework?

ConnectionFactory factory = new ConnectionFactory();
factory.setExceptionHandler(new BrokerExceptionHandler(logger, instance));

public final class BrokerExceptionHandler extends StrictExceptionHandler {
   @Override
   public void handleReturnListenerException(Channel channel, Throwable exception) {
        logger.log(Level.SEVERE, "ReturnListenerException detected: ReturnListener.handleReturn", exception);
        this.publishAlert(exception, "ReturnListener.handleReturn");
        logger.log(Level.SEVERE, "Close application", exception);
        System.exit(-1);
   }
  ....
}

Basically I need to specify a custom exception handler if a rabbitMQ exception occurs and then stop the application

How can I publish a rabbitMq message every time that there is an exception?

EDIT

I modified my configuration class in this way:

@Bean
SimpleMessageListenerContainer containerPredict(ConnectionFactory connectionFactory,
  MessageListenerAdapter listenerPredictAdapter) {
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
     container.setConnectionFactory(connectionFactory);
     container.setDefaultRequeueRejected(false);
     container.setErrorHandler(new BrokerExceptionHandler());
     container.setQueueNames(getQueueName());
     container.setMessageListener(listenerAdapter);
     return container;
}

and this is my BrokerExceptionHandler class

public class BrokerExceptionHandler implements ErrorHandler {
   private final Logger logger = Logger.getLogger(getClass().getSimpleName());

   @Autowired
   private Helper helper;

   @Override
   public void handleError(Throwable t) {
     logger.log(Level.SEVERE, "Exception Detected. Publishing error alert");
     String message = "Exception detected. Message: " + t.getMessage());
     
     // Notify the error to the System sending a new RabbitMq message
     System.out.println("---> Before convertAndSend");
     rabbitTemplate.convertAndSend(exchange, routing, message);
     System.out.println("---> After convertAndSend");
   }
}

I can see the log Exception Detected. Publishing error alert and ---> Before convertAdnSend in the console, but the new alert is not published and the log ---> After convertAndSend doesn't appear in the console.

Here it is the log:

2018-10-17 09:32:02.849 ERROR 1506 --- [tainer****-1] BrokerExceptionHandler : Exception Detected. Publishing error alert

---> Before convertAndSend

2018-10-17 09:32:02.853 INFO 1506 --- [tainer****-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@4f5b08d: tags=[{amq.ctag-yUcUmg5BCo20ucG1wJZoWA=myechange}], channel=Cached Rabbit Channel: AMQChannel(amqp://[email protected]:5672/testbed_simulators,1), conn: Proxy@3964d79 Shared Rabbit Connection: SimpleConnection@61f39bb [delegate=amqp://[email protected]:5672/testbed_simulators, localPort= 51528], acknowledgeMode=AUTO local queue size=0

2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (myexchange) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.

2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (myexchange) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.

EDIT

Debugging I see that before to send the new message, the following code is called:

File: SimpleMessageListenerContainer.class line 1212

if (!isActive(this.consumer) || aborted) {
  .....
}
else {
  ---> logger.info("Restarting " + this.consumer);
       restart(this.consumer);
}

EDIT 2

Example code: http://github.com/fabry00/spring-boot-rabbitmq

1
springspring-bootrabbitmqspring-rabbit

1 Answers

1
votes

It depends on how you are doing your configuration; if you are using Spring Boot's auto-configured connection factory...

@Bean
public InitializingBean connectionFactoryConfigurer(CachingConnectionFactory ccf) {
    return () -> ccf.getRabbitConnectionFactory().setExceptionHandler(...);
}

If you are wiring up your own beans (e.g. via a RabbitConnectionFactoryBean) then set it directly.

EDIT

You are throwing a NullPointerException in your error handler...

2018-10-17 11:51:58.733 DEBUG 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Consumer raised exception, processing can restart if the connection factory supports it

java.lang.NullPointerException: null
    at com.test.BrokerExceptionHandler.handleError(BrokerExceptionHandler.java:27) ~[main/:na]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1243) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1488) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1318) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]

2018-10-17 11:51:58.734  INFO 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@1aabf50d: tags=[{amq.ctag-VxxHKiMsWI_w8DIooAsySA=myapp.mydomain.KPIS}], channel=Cached Rabbit Channel: AMQChannel(amqp://[email protected]:5672/,1), conn: Proxy@b88a7d6 Shared Rabbit Connection: SimpleConnection@25dc64a [delegate=amqp://[email protected]:5672/, localPort= 55662], acknowledgeMode=AUTO local queue size=0

To turn on DEBUG logging, add 

logging.level.org.springframework.amqp=debug

to your application.properties.

this.helper is null because the error handler is not a Spring Bean - @Autowired only works if Spring manages the object; you are using new BrokerExceptionHandler().

EDIT2

I added these 2 beans

@Bean
public BrokerExceptionHandler errorHandler() {
    return new BrokerExceptionHandler();
}

@Bean
public MessageConverter json() { // Boot auto-configures in template
    return new Jackson2JsonMessageConverter();
}

and now...

---> Before publishing Alert event
--- ALERT
2018-10-17 12:14:45.304  INFO 43359 --- [containerKpis-1] Helper                                   : publishAlert
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.a.r.c.CachingConnectionFactory       : Creating cached Rabbit Channel from AMQChannel(amqp://[email protected]:5672/,3)
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate      : Executing callback RabbitTemplate$$Lambda$638/975724213 on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://[email protected]:5672/,3), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://[email protected]:5672/, localPort= 56220]
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate      : Publishing message (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=146, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [myevent.ALERT], routingKey = [/]
--- ALERT 2
---> After publishing Alert event
2018-10-17 12:14:45.323 DEBUG 43359 --- [pool-1-thread-6] o.s.a.r.listener.BlockingQueueConsumer   : Storing delivery for consumerTag: 'amq.ctag-eYbzZ09pCw3cjdtSprlZMQ' with deliveryTag: '1' in Consumer@4b790d86: tags=[{amq.ctag-eYbzZ09pCw3cjdtSprlZMQ=myapp.myevent.ALERT}], channel=Cached Rabbit Channel: AMQChannel(amqp://[email protected]:5672/,2), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://[email protected]:5672/, localPort= 56220], acknowledgeMode=AUTO local queue size=0
2018-10-17 12:14:45.324 DEBUG 43359 --- [ontainerReset-1] o.s.a.r.listener.BlockingQueueConsumer   : Received message: (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=myevent.ALERT, receivedRoutingKey=/, deliveryTag=1, consumerTag=amq.ctag-eYbzZ09pCw3cjdtSprlZMQ, consumerQueue=myapp.myevent.ALERT])
2018-10-17 12:14:45.324  INFO 43359 --- [ontainerReset-1] Application                              : ---> kpisAlert RECEIVED
2018-10-17 12:14:45.325 ERROR 43359 --- [ontainerReset-1] Application                              : ---> Message: Exception detected. Message: Listener method 'kpisEvent' threw exception
2018-10-17 12:14:45.326 DEBUG 43359 --- [containerKpis-1] o.s.a.r.listener.BlockingQueueConsumer   : Rejecting messages (requeue=false)

EDIT3

Or, if you prefer Gson...

  @Bean
  public MessageConverter json() {
    Gson gson = new GsonBuilder().create();
    return new MessageConverter() {

      @Override
      public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        return new Message(gson.toJson(object).getBytes(), messageProperties);
      }

      @Override
      public Object fromMessage(Message message) throws MessageConversionException {
        throw new UnsupportedOperationException();
      }

    };
  }

EDIT4

I changed the current version of your app as follows:

  @Bean
  public MessageConverter jsonConverter() {
    Gson gson = new GsonBuilder().create();
    EventKpisCollected collected = new EventKpisCollected();
    return new MessageConverter() {

      @Override
      public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        System.out.println("toMessage");
        return new Message(gson.toJson(object).getBytes(), messageProperties);
      }

      @Override
      public Object fromMessage(Message message) throws MessageConversionException {
        System.out.println("fromMessage");
        return collected.decode(new String(message.getBody()));
      }
    };
  }

...

  @Bean
  SimpleMessageListenerContainer containerKpis(ConnectionFactory connectionFactory,
      MessageListenerAdapter listenerKpisAdapter) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setDefaultRequeueRejected(false);
    container.setErrorHandler(errorHandler());
    container.setQueueNames(getQueueKpis());
    container.setMessageListener(listenerKpisAdapter);
    return container;
  }

  @Bean
  SimpleMessageListenerContainer containerReset(ConnectionFactory connectionFactory,
      MessageListenerAdapter listenerAlertAdapter) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setDefaultRequeueRejected(false);
    container.setErrorHandler(errorHandler());
    container.setQueueNames(getQueueAlert());
    container.setMessageListener(listenerAlertAdapter);
    return container;
  }

  @Bean
  MessageListenerAdapter listenerKpisAdapter(Application receiver) {
    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "kpisEvent");
    messageListenerAdapter.setMessageConverter(jsonConverter());
    return messageListenerAdapter;
  }

  @Bean
  MessageListenerAdapter listenerAlertAdapter(Application receiver) {
    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "alertEvent");
//    messageListenerAdapter.setMessageConverter(jsonConverter()); converter only handles events.
    return messageListenerAdapter;
  }

and

fromMessage
2018-10-19 13:46:53.734  INFO 10725 --- [containerKpis-1] Application                              : ---> kpisEvent RECEIVED
2018-10-19 13:46:53.734  INFO 10725 --- [containerKpis-1] Application                              : ---> kpisEvent DECODED, windowId: 1522751098000-1522752198000

With the event decoding done by the framework (just for the event currently - you will need a second converter for the alers).