NestJS Kafka生产者对接Spring Boot消费者序列化异常排查

问题场景 #

搭建NestJS Kafka生产者与Spring Boot Kafka消费者，生产者发送消息后，消费者抛出序列化相关异常。

异常日志 #

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]
    ...
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]
    ...

相关代码 #

NestJS生产者代码 #

@Injectable()
export class TestResponseService {

    constructor(private readonly producerService:ProducerService) {}

    async scheduleResponse(event:OrderEvent){
        
        console.log("Test response hits");
        await this.producerService.produce({
            topic:'schdule-topics',
            messages: [{
              //value: JSON.parse(JSON.stringify(event)),
              value: "Hi",
            }],
        });
        console.log("Test response below hits");
        return 'Hello World...!';
    }
}

Spring Boot消费者代码 #

@Service
public class Consumer{
    private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
    
    @KafkaListener(topics="${spring.kafka.topic.name}",groupId="${spring.kafka.consumer.group-id}")
    public void consume(){
        System.out.println("order ID is printing in test: ");
    }
}

解决方案 #

核心原因 #

异常根因为No type information in headers and no default type provided：Spring Kafka默认使用JsonDeserializer处理消息体，但当前NestJS发送的是普通字符串，且未提供反序列化所需的类型信息。

针对性解决 #

1. 适配普通字符串消息（当前场景）

修改Spring消费者配置，使用StringDeserializer解析消息体：

• 在application.properties中添加：

  spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  

• 同时更新消费者方法，接收字符串参数：

  @KafkaListener(topics="${spring.kafka.topic.name}",groupId="${spring.kafka.consumer.group-id}")
  public void consume(String message) {
      System.out.println("收到消息: " + message);
  }
  

2. 适配JSON对象消息（后续扩展场景）

若后续需要发送OrderEvent这类JSON对象，需保证两端序列化/反序列化逻辑一致：

• 方式一：指定默认反序列化类型
  在Spring配置中指定JSON反序列化的默认目标类：

  spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
  spring.kafka.consumer.properties.spring.json.value.default.type=com.yourpackage.OrderEvent
  

  消费者方法直接接收实体类：

  @KafkaListener(...)
  public void consume(OrderEvent event) {
      LOGGER.info("收到订单事件: {}", event);
  }
  

• 方式二：手动添加类型头
  在NestJS发送消息时，添加Spring Kafka识别的__TypeId__头，指定目标类全限定名：

  await this.producerService.produce({
      topic:'schdule-topics',
      messages: [{
          value: JSON.stringify(event),
          headers: {
              '__TypeId__': Buffer.from('com.yourpackage.OrderEvent')
          }
      }],
  });
  

3. 配置错误处理反序列化器（可选）

为避免反序列化错误导致消费者停止运行，可配置ErrorHandlingDeserializer包装实际的反序列化器：

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.apache.kafka.common.serialization.StringDeserializer

内容的提问来源于stack exchange，提问作者Anojan