title: Consume Kafka message with a custom header
body: <p>I'm trying to create a kafka Consumer with Spring Cloud Stream in order to listen a Kafka message, built outside any Spring context, with a custom header (<strong>operationType</strong>).</p>
  
  <p>I'm using Spring Boot 1.5.x / Spring Cloud Egdware.SR5 and the 1.1.1 version of kafka-client and kafka_2.11.   </p>
  
  <p>My Listener class contains this method</p>
  
  <pre><code>@StreamListener(value = "dataset-changed", condition = "headers['operationType']=='UPDATE'")
  public void onEvent(@Payload DatasetChangedMessage payload) {
    // my code should be execute only if the header operationType == UPDATE
  }
  </code></pre>
  
  <p>The Spring Cloud Stream configuration is </p>
  
  <pre><code>spring.cloud.stream:
    bindings:
      dataset-changed:
        group: preparation
        content-type: application/json
        destination: dataset-changed
        consumer:
          headerMode: raw
          configuration:
            key.deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
            value.deserializer: org.apache.kafka.common.serialization.StringDeserializer
  </code></pre>
  
  <p>The producer is a simple java doc with a kafka-client:1.1.1 library </p>
  
  <pre><code>Properties producerConfig = new Properties();
  producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
  producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  
  KafkaProducer&lt;byte[], String&gt; producer = new KafkaProducer&lt;&gt;(producerConfig);
  
  // Headers (to condition the kafka listener)
  final List&lt;Header&gt; headers = new ArrayList&lt;&gt;();
  headers.add(new RecordHeader("operationType", "UPDATE".getBytes()));
  
  ProducerRecord&lt;byte[], String&gt; record =
          new ProducerRecord&lt;&gt;("dataset-changed", 0, "111".getBytes(), getJsonPayload(), headers);
  Future&lt;RecordMetadata&gt; future = producer.send(record);
  future.get();
  
  producer.close();
  </code></pre>
  
  <p>When i produce the kafka message, i have this kind of warnings</p>
  
  <pre><code>2019-03-15 14:48:32.103  WARN [tdp-preparation,1e24b9764ef9bb14,1e24b9764ef9bb14,false] 34760 --- [           -L-1] .DispatchingStreamListenerMessageHandler : Cannot find a @StreamListener matching for message with id: ea27a446-69da-7b8d-1b94-50b46a40dfde
  </code></pre>
  
  <p>whereas the operationType header is present </p>
  
  <p><a href="https://i.stack.imgur.com/P6XhD.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/P6XhD.png" alt="enter image description here"></a></p>