title: Error handling in Spring RabbitMQ
body: <p>I am trying to integrate Rabbit MQ broker in our Spring application. I am able to consume messages successfully but need to add Error handling. 
  Listener consumes the message and apply business logic to it, which include DB writes. The Business logic can throw exception. </p>
  
  <p>In case of these exceptions I need to </p>
  
  <ol>
  <li>Rollback the Db writes.</li>
  <li>Write to error table in Db, indicating msg failure.</li>
  <li>Message should not be re-queued.</li>
  </ol>
  
  <p>For </p>
  
  <ul>
  <li><p>requirement #1 - have added <code>txManager</code> in <code>config.xml</code> and annotated the <code>Listner.listen()</code> method with <code>@Transactional</code></p></li>
  <li><p>requirement #2 - have added Error handler and custom implementation of <code>DefaultExceptionStrategey</code></p></li>
  <li><p>requirement #3 - have set <code>DefaultRequeueRejected=false</code></p></li>
  </ul>
  
  <p>But when <code>BusinessRuntimeException</code> is thrown from Listener, ErrorHandler is not getting invoked. 
  Don't know what is missing.
  Is <code>errorHandler</code> invoked only for some Exceptions?</p>
  
  <p><strong>Config.xml</strong></p>
  
  <pre><code>&lt;tx:annotation-driven transaction-manager="txManager" /&gt;
  &lt;bean id="txManager"
  class="org.springframework.transaction.jta.JtaTransactionManager"&gt;
  &lt;property name="allowCustomIsolationLevels" value="true" /&gt;
  </code></pre>
  
  <p></p>
  
  <pre><code>&lt;rabbit:connection-factory id="rabbitConnectionFactory"/&gt;
  &lt;rabbit:template id="rabbitTemplate" connection-
  factory="rabbitConnectionFactory" message-converter="jsonMessageConverter" 
  channel-transacted="true"/&gt;
  &lt;rabbit:admin id="rabbitAdmin" connection-factory="rabbitConnectionFactory"/&gt; 
  </code></pre>
  
  <p><strong>RabbitMQConfiguration.java</strong></p>
  
  <pre><code>@Configuration
  @EnableRabbit
  public class RabbitMqConfiguration {
  
  @Autowired
  private ConnectionFactory rabbitConnectionFactory;
  
  @Autowired
  private MessageConverter jsonMessageConverter;
  
  @Bean
  public SimpleRabbitListenerContainerFactory exportPartyListenerContainer() {
      SimpleRabbitListenerContainerFactory listenerContainer = new SimpleRabbitListenerContainerFactory();
      listenerContainer.setConnectionFactory(rabbitConnectionFactory);
      listenerContainer.setMessageConverter(jsonMessageConverter);
      listenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
      listenerContainer.setChannelTransacted(true);
      listenerContainer.setDefaultRequeueRejected(false);
      listenerContainer.setErrorHandler(errorHandler());
      return listenerContainer;
  }
  
  @Bean
  public ErrorHandler errorHandler() {
      return new ConditionalRejectingErrorHandler(new ExceptionStrategy());
  } }   
  </code></pre>
  
  <p><strong>ExceptionStrategy.java</strong></p>
  
  <pre><code>public class ExceptionStrategy extends DefaultExceptionStrategy {
  
  @Autowired
  private Dao daoBean;
  
  @Override
  public boolean isFatal(Throwable t) {
  
      if (t instanceof BusinessRuntimeException) {
          BusinessRuntimeException businessException = (BusinessRuntimeException) t;
          //db call
          daoBean.updateRecordStaus();
          return true;
      }
  
      if (t instanceof ListenerExecutionFailedException) {
          ListenerExecutionFailedException lefe = (ListenerExecutionFailedException) t;
          logger.error(
                  "Failed to process inbound message from queue " + lefe.getFailedMessage().getMessageProperties().getConsumerQueue()
                          + "; failed message: " + lefe.getFailedMessage(),
                  t);
      }
      return super.isFatal(t);
  }}   
  </code></pre>