Spring应用中Kafka消息首次无法消费，多次重试后才处理的问题

问题描述 #

使用Spring应用向Kafka Topic发送JSON Payload，生产者端发送消息成功，但消费者首次无法处理该消息，经过多次重试后，同一条消息才被处理成功。期望消息能被消费者正常消费，使用的Kafka版本为2.13。

生产者代码 #

KafkaTemplate<String, Object> kafkaTemplate;

public void producer(JsonPayload payLoad) {
    try {
        kafkaTemplate.send(kafkaTopic, payLoad);
    } catch (Exception e) {
        throw e;
    }
}

消费者代码 #

@KafkaListener(topics = "${kafka.topics") // 注意此处存在语法错误：缺少闭合的}
public void listener(Object payLoad) {
    if (payLoad != null) {
        try {
            service.processPayload(payLoad);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

可能的原因及解决方案 #

1. 消费者注解语法错误 #

消费者代码中@KafkaListener的配置项${kafka.topics}缺少闭合的}，会导致配置加载异常，引发消费者初始化问题，进而导致首次消费失败。
解决： 修正注解配置：

@KafkaListener(topics = "${kafka.topics}")

2. JSON反序列化配置缺失 #

生产者发送的是JSON格式消息，但消费者未配置对应的JSON反序列化器，直接用Object接收会导致反序列化失败，重试时可能因类加载或其他偶然因素成功。
解决：

• 配置生产者使用JSON序列化器：
  在application.properties/yaml中添加：

  spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
  

• 配置消费者使用JSON反序列化器：

  spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
  spring.kafka.consumer.properties.spring.json.trusted.packages=* # 替换为你的实体类包名更安全
  

• 消费者方法直接指定接收类型，避免用Object：

  public void listener(JsonPayload payLoad) {
      // ...
  }
  

3. 业务方法依赖未就绪 #

service.processPayload方法内部依赖的服务可能存在延迟初始化，首次调用时依赖未就绪导致失败，重试时依赖服务已启动完成。
解决：

• 检查processPayload内部逻辑，确保所有依赖服务在消费者启动前完成初始化；
• 若依赖是异步初始化，可在消费者方法中添加等待逻辑或依赖健康检查机制。

4. 日志排查不足 #

消费者异常仅通过e.printStackTrace()打印，无法留存完整的首次失败日志，不利于排查问题。
解决： 使用日志框架（如SLF4J）记录详细异常信息：

private static final Logger log = LoggerFactory.getLogger(当前类.class);

// ...
catch (Exception e) {
    log.error("处理消息失败", e);
}

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