title: How to manually acknowledge RabbitMQ messages in Spring Cloud Stream?
body: <p>For stream-based services, I want the message to remain in the queue when the underlying service invoked within a <code>@StreamListener</code> fails. To that end, my understanding is that the only way to do this is to configure <code>spring.cloud.stream.bindings.channel_name.consumer.acknowledge-mode=MANUAL</code>.</p>
  
  <p>After making this configuration change, I tried adding <code>@Header(AmqpHeaders.CHANNEL) Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag</code> as method arguments to my existing <code>@StreamListener</code> implementation as documened at <a href="https://docs.spring.io/spring-integration/reference/html/amqp.html#amqp-inbound-ack" rel="nofollow noreferrer">https://docs.spring.io/spring-integration/reference/html/amqp.html#amqp-inbound-ack</a>. With this code in place, I encountered the following exception:</p>
  
  <pre><code>org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
      at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:941)
      at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:851)
      at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:771)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:102)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:198)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1311)
      at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:752)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1254)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1224)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:102)
      at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1470)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.springframework.messaging.MessageHandlingException: Missing header 'amqp_channel' for method parameter type [interface com.rabbitmq.client.Channel]
      at org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver.handleMissingValue(HeaderMethodArgumentResolver.java:100)
      at org.springframework.messaging.handler.annotation.support.AbstractNamedValueMethodArgumentResolver.resolveArgument(AbstractNamedValueMethodArgumentResolver.java:103)
      at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:112)
  </code></pre>
  
  <p>I then found the following: <a href="https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_usage_examples" rel="nofollow noreferrer">https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_usage_examples</a>, which shows an example of how to perform acknowledgement of messages using Kafka, but I am currently using the RabbitMQ binding. We plan on eventually moving to Kafka, but for now, how do I configure and code a solution to do manual message acknowledgement for successfully processed messages and manual message rejection, thus leaving the message on the queue, when exceptions are encountered. I am currently on Spring Cloud <code>Edgware.RELEASE</code> and Spring Cloud Stream <code>Ditmars.RELEASE</code>.</p>
  
  <p><strong>UPDATE</strong></p>
  
  <p>Now I have the following configuration:</p>
  
  <pre><code>spring:
    cloud:
      stream:
        bindings:
          do-something-async-reply:
            group: xyz-service-do-something-async-reply
        rabbit:
          bindings:
            do-something-async-reply:
              consumer:
                autoBindDlq: true
                dlqDeadLetterExchange:
                dlqTtl: 10000
                requeueRejected: true
  </code></pre>
  
  <p>And I'm receiving the following error at service startup:</p>
  
  <pre><code>2018-01-12 14:46:34.346 ERROR [xyz-service,,,] 2488 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method&lt;channel.close&gt;(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'do-something-async-reply.xyz-service-do-something-async-reply' in vhost '/': received the value 'DLX' of type 'longstr' but current is none, class-id=50, method-id=10)
  </code></pre>
  
  <p>What configuration is wrong/am I missing?</p>