title: Spring AMQP - Message re queuing using dead letter mechanism with TTL
body: <p>Its like "<strong>Houston we have a problem here</strong>" where I need to schedule/delay a message for 5 minutes after it fails on the first attempt to process an event.
  I have implemented dead letter exchange in this scenario.</p>
  
  <p>The messages on failing, route to the DLX --> Retry Queue and comes back to work queue after a TTL of 5 minutes for another attempt.</p>
  
  <p>Here is the configuration I am using:</p>
  
  <pre><code>public class RabbitMQConfig {
      @Bean(name = "work")
      @Primary
      Queue workQueue() {
          return new Queue(WORK_QUEUE, true, false, false, null);
      }
  
      @Bean(name = "workExchange")
      @Primary
      TopicExchange workExchange() {
          return new TopicExchange(WORK_EXCHANGE, true, false);
      }
  
      @Bean
      Binding workBinding(Queue queue, TopicExchange exchange) {
          return BindingBuilder.bind(workQueue()).to(workExchange()).with("#");
      }
  
      @Bean(name = "retryExchange")
      FanoutExchange retryExchange() {
          return new FanoutExchange(RETRY_EXCHANGE, true, false);
      }
  
      @Bean(name = "retry")
      Queue retryQueue() {
          Map&lt;String, Object&gt; args = new HashMap&lt;String, Object&gt;();
          args.put("x-dead-letter-exchange", WORK_EXCHANGE);
          args.put("x-message-ttl", RETRY_DELAY); //delay of 5 min
          return new Queue(RETRY_QUEUE, true, false, false, args);
      }
  
      @Bean
      Binding retryBinding(Queue queue,FanoutExchange exchange) {
          return BindingBuilder.bind(retryQueue()).to(retryExchange());
      }
  
      @Bean
      public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
          SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
          factory.setConnectionFactory(connectionFactory);
          return factory;
      }
  
      @Bean
      Consumer receiver() {
          return new Consumer();
      }
  
      @Bean
      MessageListenerAdapter listenerAdapter(Consumer receiver) {
          return new MessageListenerAdapter(receiver, "receiveMessage");
      }
  }
  </code></pre>
  
  <p>Producer.java:</p>
  
  <pre><code>@GetMapping(path = "/hello")
  public String sayHello() {
      // Producer operation
  
          String messages[];
          messages = new String[] {" hello "};
  
      for (int i = 0; i &lt; 5; i++) {
          String message = util.getMessage(messages)+i;
  
          rabbitTemplate.convertAndSend("WorkExchange","", message);
         System.out.println(" Sent '" + message + "'");
      }
      return "hello";
  }
  </code></pre>
  
  <p>Consumer.java:</p>
  
  <pre><code>public class Consumer {
  
      @RabbitListener(queues = "WorkQueue")
      public void receiveMessage(String message, Channel channel,
              @Header(AmqpHeaders.DELIVERY_TAG) Long tag) throws IOException, InterruptedException {
  
          try {
  
              System.out.println("message to be processed: " + message);
              doWorkTwo(message);
              channel.basicAck(tag, false);
  
          } catch (Exception e) {
              System.out.println("In the exception catch block");
              System.out.println("message in dead letter exchange: " + message);
              channel.basicPublish("RetryExchange", "", null, message.getBytes());
  
          }
  
      }
  
      private void doWorkTwo(String task) throws InterruptedException {
  
          int c = 0;
          int b = 5;
          int d = b / c;
  
      }
  
  }
  </code></pre>
  
  <p>Is it the correct way to use a dead letter exchange for my scenario and <strong>after waiting once in the RETRY QUEUE for 5 min, on the second time attempt it does not wait for 5 min in the RETRY QUEUE (I have mentioned TTL as 5 min) and moves to the WORK QUEUE immediately</strong>.</p>
  
  <p>I am running this application by hitting localhost:8080/hello url. </p>
  
  <p><strong><em>Here is my updated configuration.</em></strong></p>
  
  <p>RabbitMQConfig.java:</p>
  
  <pre><code>@EnableRabbit
  public class RabbitMQConfig {
  
      final static String WORK_QUEUE = "WorkQueue";
      final static String RETRY_QUEUE = "RetryQueue";
      final static String WORK_EXCHANGE = "WorkExchange"; // Dead Letter Exchange
      final static String RETRY_EXCHANGE = "RetryExchange";
      final static int RETRY_DELAY = 60000; // in ms (1 min)
  
      @Bean(name = "work")
      @Primary
      Queue workQueue() {
           Map&lt;String, Object&gt; args = new HashMap&lt;String, Object&gt;();
           args.put("x-dead-letter-exchange", RETRY_EXCHANGE);
          return new Queue(WORK_QUEUE, true, false, false, args);
      }
  
      @Bean(name = "workExchange")
      @Primary
      DirectExchange workExchange() {
          return new DirectExchange(WORK_EXCHANGE, true, false);
      }
  
      @Bean
      Binding workBinding(Queue queue, DirectExchange exchange) {
          return BindingBuilder.bind(workQueue()).to(workExchange()).with("");
      }
  
      @Bean(name = "retryExchange")
      DirectExchange retryExchange() {
          return new DirectExchange(RETRY_EXCHANGE, true, false);
      }
  
      // Messages will drop off RetryQueue into WorkExchange for re-processing
      // All messages in queue will expire at same rate
      @Bean(name = "retry")
      Queue retryQueue() {
          Map&lt;String, Object&gt; args = new HashMap&lt;String, Object&gt;();
          //args.put("x-dead-letter-exchange", WORK_EXCHANGE);
          //args.put("x-message-ttl", RETRY_DELAY);
          return new Queue(RETRY_QUEUE, true, false, false, null);
      }
  
      @Bean
      Binding retryBinding(Queue queue, DirectExchange exchange) {
          return BindingBuilder.bind(retryQueue()).to(retryExchange()).with("");
      }
  
      @Bean
      public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
          SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
          factory.setConnectionFactory(connectionFactory);
          factory.setDefaultRequeueRejected(false);
          /*factory.setAdviceChain(new Advice[] {
                  org.springframework.amqp.rabbit.config.RetryInterceptorBuilder
                          .stateless()
                          .maxAttempts(2).recoverer(new RejectAndDontRequeueRecoverer())
                          .backOffOptions(1000, 2, 5000)
                          .build()
          });*/
          return factory;
      }
  
      @Bean
      Consumer receiver() {
          return new Consumer();
      }
  
      @Bean
      MessageListenerAdapter listenerAdapter(Consumer receiver) {
          return new MessageListenerAdapter(receiver, "receiveMessage");
      }
  }
  </code></pre>
  
  <p>Consumer.java:</p>
  
  <pre><code>public class Consumer {
  
      @RabbitListener(queues = "WorkQueue")
      public void receiveMessage(String message, Channel channel,
              @Header(AmqpHeaders.DELIVERY_TAG) Long tag,
              @Header(required = false, name = "x-death") HashMap&lt;String, String&gt; xDeath)
              throws IOException, InterruptedException {
  
          doWorkTwo(message);
          channel.basicAck(tag, false);
      }
  
      private void doWorkTwo(String task) {
          int c = 0;
          int b = 5;
          if (c &lt; b) {
              throw new AmqpRejectAndDontRequeueException(task);
          }
      }
  }
  </code></pre>