Spring Cloud Stream消费Kafka Topic无效JSON反序列化异常触发问题

问题根因 #

• Jackson默认配置下会忽略JSON尾部的不匹配冗余字符，你输入的{}}}会被Jackson截断为合法的空JSON对象{}，反序列化过程不会抛出异常，因此不会触发ErrorHandlingDeserializer的错误处理逻辑，直接进入业务消费方法。
• @JsonProperty(required = true)以及Lombok的@NonNull注解默认不会在Jackson反序列化阶段生效：前者只对序列化逻辑生效，后者仅作用于你自己手动调用的构造方法/Setter方法，Jackson通过反射调用Setter时不会触发@NonNull校验，因此空JSON对象反序列化后得到的是所有属性为null的KafkaEventRecord对象。
• 你观察到的错误方法触发2次是Spring Cloud Stream默认的消费重试机制导致，默认最大消费尝试次数为3，因此首次消费失败后会重试2次，属于预期行为。

解决方案 #

1. 配置Jackson反序列化规则，非法JSON直接抛出异常 #

自定义ObjectMapper配置，开启尾部冗余字符校验，避免不完整JSON被误判为合法对象：

@Configuration
public class JacksonConfig {
    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper objectMapper = new ObjectMapper();
        // 遇到JSON尾部有多余字符时抛出异常
        objectMapper.enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
        // 反序列化时遇到必填字段缺失抛出异常
        objectMapper.enable(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES);
        objectMapper.enable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES);
        return objectMapper;
    }
}

同时修改KafkaEventRecord类，移除无参构造和Setter，强制Jackson使用全参构造反序列化，让@JsonProperty(required = true)生效：

@Builder
@Getter
@AllArgsConstructor
public class KafkaEventRecord {
    @JsonProperty(value = "transport_metadata", required = true)
    private TransportMetadata transportMetadata;

    @JsonProperty(value = "payload", required = true)
    private Payload payload;
}

2. 配置JsonDeserializer使用自定义的ObjectMapper #

默认的Spring Kafka JsonDeserializer不会使用Spring容器里的ObjectMapper，需要手动绑定，在配置文件中添加如下配置：

spring.cloud.stream.kafka.bindings.consumer-in-0.consumer.configuration.spring.json.value.default.type=你的项目全限定类名.KafkaEventRecord
spring.cloud.stream.kafka.bindings.consumer-in-0.consumer.configuration.spring.json.object.mapper=com.fasterxml.jackson.databind.ObjectMapper

3. 可选：添加消费前参数校验 #

如果不想修改类结构，可以在消费逻辑前手动校验参数，异常会被框架捕获自动转发到errorChannel：

@Bean
Consumer<Message<KafkaEventRecord>> consumer(Validator validator) {
    return message -> {
        Set<ConstraintViolation<KafkaEventRecord>> violations = validator.validate(message.getPayload());
        if (!violations.isEmpty()) {
            throw new ConstraintViolationException(violations);
        }
        log.info("Kafka Event data consumed from Kafka {}", message.getPayload());
    };
}

4. 可选：调整消费重试次数 #

如果不需要重试，可以修改配置关闭重试：

spring.cloud.stream.bindings.consumer-in-0.consumer.max-attempts=1

内容的提问来源于stack exchange，提问作者Batıkan Türkmen