title: Spring cloud stream: Headers kafka_acknowledgment value is null
body: <p>I am trying to have the control over the offset commit on kafka topic according to the status of processing of message in the application. If message are successful, so offset can be committed. For this I am trying to get the Headers in my method if so that message can be acknowledge manually </p>
  
  <pre><code>  spring:
     cloud:
      stream:
        default:
          contentType: application/json
        default-binder: binder1-kafka
        bindings:
          myChannel:
            binder: binder1-kafka
            destination: my_topic
            content-type: text/plain
            consumer:
              autoCommitOffset: false
  
          outChannel:
            binder: binder2-kafka
            destination: my_topic
            content-type: text/plain
            consumer:
              autoCommitOffset: false    
  
  
        binders:
          #Connection config to different clusters
          binder1-kafka:
            type: kafka
            defaultCandidate: true
            environment:
              spring:
                cloud:
                  stream:
                    kafka:
                      binder:
                        brokers: some-url1:9092
          binder2-kafka:
            type: kafka
            defaultCandidate: false
            environment:
              spring:
                cloud:
                  stream:
                    kafka:
                      binder:
                        brokers: some-url2:9092 
  </code></pre>
  
  <p>But When using the listener as  </p>
  
  <pre><code>@StreamListener(target = IBrokerChannel.myChannel )
  public void handlePayload(@Payload MyPayload payload, @Headers Map&lt;String, Object&gt; headers) {
    Acknowledgment acknowledgment= (Acknowledgment) headers.get("kafka_acknowledgment"); // acknowledgment object is always null.
    acknowledgment.acknowledge();
  }
  </code></pre>
  
  <p><code>acknowledgment</code> is always <code>null</code>. I am using kafka producer cli to send the message to the topic. <code>spring-boot</code> version is <code>1.5.10.RELEASE</code></p>