title: spring kafka No type information in headers and no default type provided
body: <p>I've got a spring boot app that defines: </p>
  
  <ul>
  <li>a REST Controller that writes to a kafka topic, STREAM_TOPIC_IN_QQQ</li>
  <li>a KafkaListener that reads from STREAM_TOPIC_IN_QQQ (groupId="bar") and logs</li>
  <li>a KStream that peeks the topic and logs it, converts it to another type, then writes it to STREAM_TOPIC_OUT_QQQ</li>
  <li>another KafkaListener that reads from STREAM_TOPIC_OUT_QQQ.</li>
  </ul>
  
  <p>(I've been changing the suffix to avoid any possible confusion, and creating the topics by hand, because otherwise I was getting a warning, STREAM_TOPIC_IN_<em>xxx</em>=LEADER_NOT_AVAILABLE and the stream would not run for a minute or so.)</p>
  
  <p>The first listener and the stream seem to be working, but when the listener on the STREAM_OUT_TOPIC tries to deserialize the message, I get the exception below.  I am providing the serde in the stream with Produced.with.  What do I need to do so that the listener knows the type to deserialize to?</p>
  
  <p><strong>Log</strong></p>
  
  <pre><code>11 Mar 2019 14:34:00,194   DEBUG    [KafkaMessageController [] http-nio-8080-exec-1]   Sending a Kafka Message
  11 Mar 2019 14:34:00,236   INFO     [KafkaConfig [] kafka9000-v0.1-b0a60795-0258-48d9-8c87-30fa9a97d7b8-StreamThread-1]   -------------- STREAM_IN_TOPIC peek: Got a greeting in the stream: Hello, World!
  11 Mar 2019 14:34:00,241   INFO     [KafkaConfig [] org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1]   STREAM_IN_TOPIC Listener: ConsumerRecord: {}ConsumerRecord(topic = STREAM_TOPIC_IN_QQQ, partition = 0, offset = 0, CreateTime = 1552332840188, serialized key size = 1, serialized value size = 34, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1, value = com.teramedica.kafakaex001web.model.Greeting@7b6c8fcc)
  11 Mar 2019 14:34:00,243   INFO     [Metadata [] kafka-producer-network-thread | kafka9000-v0.1-b0a60795-0258-48d9-8c87-30fa9a97d7b8-StreamThread-1-producer]   Cluster ID: y48IEZaGQWKcWDVGf4mD6g
  11 Mar 2019 14:34:00,367   ERROR    [LoggingErrorHandler [] org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]   Error while processing: ConsumerRecord(topic = STREAM_TOPIC_OUT_QQQ, partition = 0, offset = 0, CreateTime = 1552332840188, serialized key size = 1, serialized value size = 48, headers = RecordHeaders(headers = [RecordHeader(key = springDeserializerExceptionValue, value = [ REDACTED ])], isReadOnly = false), key = 1, value = null)
  org.springframework.kafka.support.serializer.DeserializationException: failed to deserialize; nested exception is java.lang.IllegalStateException: No type information in headers and no default type provided
      at org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2.deserializationException(ErrorHandlingDeserializer2.java:204) ~[spring-kafka-2.2.4.RELEASE.jar:2.2.4.RELEASE]
  </code></pre>
  
  <p>Here's the configuration:</p>
  
  <p><strong>REST (spring mvc):</strong></p>
  
  <pre><code>@RequestMapping("/greeting")
  public Greeting greeting(@RequestParam(value = "name", defaultValue = "World") String name) {
      Greeting gr = new Greeting(counter.incrementAndGet(), String.format(msgTemplate, name));
      this.kafkaTemplate.send(K9000Consts.STREAM_TOPIC_IN, "1", gr);
      logger.debug("Sending a Kafka Message");
      return gr;
  }
  </code></pre>
  
  <p><strong>Kafka Config (spring-kafka):</strong></p>
  
  <pre><code>@Bean
  public KStream&lt;String, Greeting&gt; kStream(StreamsBuilder kStreamBuilder) {
      KStream&lt;String, Greeting&gt; stream = kStreamBuilder.stream(K9000Consts.STREAM_TOPIC_IN);
      stream.peek((k, greeting) -&gt; {
          logger.info("-------------- STREAM_IN_TOPIC peek: Got a greeting in the stream: {}", greeting.getContent());
      })
            .map((k, v) -&gt; new KeyValue&lt;&gt;(k, new GreetingResponse(v)))
            .to(K9000Consts.STREAM_TOPIC_OUT, Produced.with(stringSerde, new JsonSerde&lt;&gt;(GreetingResponse.class)));
      return stream;
  }
  
  @KafkaListener(topics = K9000Consts.STREAM_TOPIC_OUT, groupId="oofda", errorHandler = "myTopicErrorHandler")
  public void listenForGreetingResponse(ConsumerRecord&lt;String, GreetingResponse&gt; cr) throws Exception {
      logger.info("STREAM_OUT_TOPIC Listener : {}" + cr.toString());
  }
  
  @KafkaListener(topics = K9000Consts.STREAM_TOPIC_IN, groupId = "bar")
  public void listenForGreetingResponses(ConsumerRecord&lt;String, Greeting&gt; cr) throws Exception {
      logger.info("STREAM_IN_TOPIC Listener: ConsumerRecord: {}" + cr.toString());
  }
  </code></pre>
  
  <p>application.yml</p>
  
  <pre><code>spring:
  kafka:
    bootstrap-servers:  localhost:9092
    consumer:
      group-id: foo
      auto-offset-reset: latest
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2
      properties:
        spring.json.trusted.packages: com.teramedica.kafakaex001web.model
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    streams:
      application-id: kafka9000-v0.1
      properties: # properties not explicitly handled by KafkaProperties.streams
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.springframework.kafka.support.serializer.JsonSerde
        spring.json.trusted.packages: com.teramedica.kafakaex001web.model
  </code></pre>