title: Spring kafka message deserialization fails because the content-type is set to application/json
body: <p>I am sending a message to Kafka as follows:</p>
  
  <pre><code>private KafkaTemplate&lt;String, MyMessage &gt; kafkaTemplate;
  
  public void sendMessage(MyMessage data) {
  
      Message&lt;MyMessage &gt; message = MessageBuilder
              .withPayload(data)
              .setHeader(KafkaHeaders.TOPIC,topic)
              .setHeader(KafkaHeaders.MESSAGE_KEY,data.getKey())
              .build();
      kafkaTemplate.send(message);
  </code></pre>
  
  <p>I have my own custom de/serializer for MyMessage, The key is a simple String</p>
  
  <p>I am trying to process the message as follows:</p>
  
  <pre><code>@StreamListener
      public void process(@Input("input") KStream&lt;String,MyMessage&gt; myStream){
              final Serde&lt;String&gt; stringSerde = Serdes.String();
              final MyMessageSerde myMessageSerde = new MyMessageSerde();
              myStream
                      .groupBy((key,value)-&gt; value.getObjectKey(), Serialized.with(stringSerde,myMessageSerde))
                      .aggregate(ArrayList::new,
                              (newKey,val,agg) -&gt; {
                                  agg.add(val);
                                  return agg;
                              },
                              Materialized.&lt;String, ArrayList&lt;MyMessage&gt;, KeyValueStore&lt;Bytes, byte[]&gt;&gt;as("object-keys")
                                      .withKeySerde(stringSerde)
                                      .withValueSerde(new ArrayListSerde(myMessageSerde)));
  
      ...
      interface MyStreamProcessor {
          @Input("input")
          KStream&lt;?, ?&gt; input();
      }
  </code></pre>
  
  <p>This is failing with a deserialization error and after debugging I see the reason is that the content-type header of the message is set to application/json and therefore it is trying to deserialize the message as JSON. How does the content-type header get set and how can I override it? I am not sending the messages as JSON but rather as a byte[] like regular Kafka messages so I want my custom de/serializer to be used</p>
  
  <p>What is even stranger is that if I change this to a KTable it works fine.</p>
  
  <pre><code>@StreamListener
  public void process(@Input("input") KTable&lt;String,MyMessage&gt; myTable){
          final Serde&lt;String&gt; stringSerde = Serdes.String();
          final MyMessageSerde myMessageSerde = new MyMessageSerde();
          KStream&lt;String,MyMessage&gt; myStream = myTable.toStream();
          myStream
                  .groupBy((key,value)-&gt; value.getObjectKey(), Serialized.with(stringSerde,myMessageSerde))
                  .aggregate(ArrayList::new,
                          (newKey,val,agg) -&gt; {
                              agg.add(val);
                              return agg;
                          },
                          Materialized.&lt;String, ArrayList&lt;MyMessage&gt;, KeyValueStore&lt;Bytes, byte[]&gt;&gt;as("object-keys")
                                  .withKeySerde(stringSerde)
                                  .withValueSerde(new ArrayListSerde(myMessageSerde)));
  
  ...
  interface MyStreamProcessor {
      @Input("input")
      KTable&lt;?, ?&gt; input();
  }
  </code></pre>
  
  <p>Why with a KTable is the message deserialized correctly? What is the difference in the behavior between a KStream and a KTable?</p>