title: NestJs Kafka to Spring Kafka
body: <p>I Have Simple NestJs Kafka producer and Springboot Kafka Consumer . when i send messge in kafka producer the kafka consumer give the following error.</p>
  <p>//</p>
  <pre><code>java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
      at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:151) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1809) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1301) ~[spring-kafka-2.8.8.jar:2.8.8]
      at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[na:na]
      at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
      at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
  Caused by: org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition schdule-topics-0 at offset 1. If needed, please seek past the record to continue consumption.
      at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1448) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.internals.Fetcher.access$3400(Fetcher.java:135) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1671) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1900(Fetcher.java:1507) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:733) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:684) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1277) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) ~[kafka-clients-3.1.1.jar:na]
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) ~[kafka-clients-3.1.1.jar:na]
      at jdk.internal.reflect.GeneratedMethodAccessor50.invoke(Unknown Source) ~[na:na]
      at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
      at java.base/java.lang.reflect.Method.invoke(Method.java:568) ~[na:na]
      at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.3.22.jar:5.3.22]
      at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208) ~[spring-aop-5.3.22.jar:5.3.22]
      at jdk.proxy2/jdk.proxy2.$Proxy112.poll(Unknown Source) ~[na:na]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1529) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1519) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1343) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1255) ~[spring-kafka-2.8.8.jar:2.8.8]
      ... 3 common frames omitted
  Caused by: java.lang.IllegalStateException: No type information in headers and no default type provided
      at org.springframework.util.Assert.state(Assert.java:76) ~[spring-core-5.3.22.jar:5.3.22]
      at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:583) ~[spring-kafka-2.8.8.jar:2.8.8]
      at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1439) ~[kafka-clients-3.1.1.jar:na]
      ... 21 common frames omitted
  
  </code></pre>
  <p>// Here is my codes</p>
  <p>@Injectable()
  export class TestResponseService {</p>
  <pre><code>constructor(private readonly producerService:ProducerService) {}
  </code></pre>
  <p>//</p>
  <pre><code>async scheduleResponse(event:OrderEvent){
      
      console.log(&quot;Test response hits&quot;);
      await this.producerService.produce({
          topic:'schdule-topics',
          messages: [{
            //value: JSON.parse(JSON.stringify(event)),
            value: &quot;Hi&quot;,
  
         
          
          },],
      });
      console.log(&quot;Test response below hits&quot;);
  
      
      return 'Hello World...!';
      
  }
  </code></pre>
  <p>}</p>
  <p>// and My Spring boot code here</p>
  <pre><code>@Service
  public class Consumer{
      
      
      
      private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
      
      @KafkaListener(topics=&quot;${spring.kafka.topic.name}&quot;,groupId=&quot;${spring.kafka.consumer.group-id}&quot;)
      
      public void consume(){
          //LOGGER.info(String.format(&quot;Order event received in Test =&gt; %s&quot;, event.toString()));
          
          System.out.println(&quot;order ID is printing in test: &quot;);
          
          
          
      }
      
  
  }
  </code></pre>