Java Kafka客户端序列化LocalDateTime异常：注解与自定义序列化器失效

问题描述 #

需要将以下POJO序列化为JSON后发送至Kafka：

@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MyObject {
    @NonNull
    @JsonProperty(required = true)
    @JsonSerialize(using = LocalDateTimeSerializer.class)
    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
    LocalDateTime eventTime;
}

但序列化后的JSON中eventTime字段始终为数组格式，而非预期的格式化字符串：

{
    "eventTime": [
        2023,
        1,
        30,
        8,
        57,
        57,
        248107000
    ]
}

已尝试以下操作，但均无效：

1. 创建自定义序列化器：

public class JRS310Serializer extends JsonSerializer<MyObject> {
    public JRS310Serializer() {
        super();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
    }
}

2. 在application.yaml中配置该序列化器：

spring:
  kafka:
    producer:
      value-serializer: org.package.JRS310Serializer 

3. 在application.yaml中添加Jackson全局配置：

spring:
  jackson:
    serialization:
      WRITE_DATES_AS_TIMESTAMPS: false

调试发现自定义序列化器中的ObjectMapper配置正确，直接调用writeValueAsString能得到正常JSON，但Kafka实际使用ObjectWriter的writeValueAsBytes序列化时仍输出数组格式，Spring Boot 2.7.7和2.7.8版本均存在此问题。

问题原因 #

你自定义的JRS310Serializer继承的是Jackson的com.fasterxml.jackson.databind.JsonSerializer，但Spring Kafka的value-serializer要求的是实现org.apache.kafka.common.serialization.Serializer接口的类，两者属于不同的抽象体系，因此你的自定义序列化器并未被Kafka正确调用。

另外，Spring Kafka默认的JsonSerializer不会自动复用Spring Boot全局配置的ObjectMapper，需要显式指定配置逻辑才能生效。

解决方案 #

方案一：自定义Kafka Serializer实现类 #

实现Kafka标准的Serializer接口，使用配置好的ObjectMapper完成序列化：

public class JRS310KafkaSerializer implements Serializer<MyObject> {
    private final ObjectMapper objectMapper;

    public JRS310KafkaSerializer() {
        this.objectMapper = new ObjectMapper()
                .registerModule(new JavaTimeModule())
                .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
    }

    @Override
    public byte[] serialize(String topic, MyObject data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Failed to serialize MyObject to JSON", e);
        }
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 可根据需求添加配置初始化逻辑
    }

    @Override
    public void close() {
        // 资源清理操作
    }
}

然后在application.yaml中配置该序列化器：

spring:
  kafka:
    producer:
      value-serializer: org.package.JRS310KafkaSerializer

方案二：复用Spring Kafka默认JsonSerializer并注入自定义ObjectMapper #

1. 创建配置类，生成配置好的ObjectMapper和JsonSerializer Bean：

@Configuration
public class JacksonConfig {
    @Bean
    public ObjectMapper objectMapper() {
        return new ObjectMapper()
                .registerModule(new JavaTimeModule())
                .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
    }

    @Bean
    public JsonSerializer<MyObject> jsonSerializer(ObjectMapper objectMapper) {
        return new JsonSerializer<>(objectMapper);
    }
}

2. 在application.yaml中配置使用Spring Kafka的默认JsonSerializer：

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

此时Spring会自动将配置好的ObjectMapper注入到JsonSerializer中，实现日期的格式化输出。

方案三：通过配置参数直接修改Spring Kafka的JsonSerializer行为 #

无需编写额外代码，直接通过application.yaml配置参数启用JavaTime模块并关闭日期时间戳序列化：

spring:
  kafka:
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        spring.json.add.type.headers: false
        spring.json.format.write_dates_as_timestamps: false
        spring.json.format.register_jdk8_module: true

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