我需要发送给Kafka简森从以下pojo连载:
@Builder
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MyObject {
@NonNull
@JsonProperty(required = true)
@JsonSerialize(using = LocalDateTimeSerializer.class)
@JsonDeserialize(using = LocalDateTimeDeserializer.class)
LocalDateTime eventTime;
}
不幸的是,即使有注解,我也看到在结果json中我得到了以下数据,而不是格式化字符串:
{
"eventTime": [
2023,
1,
30,
8,
57,
57,
248107000
]
}
根据类似问题的其他答案,我尝试了几件事:我创建了自定义序列化程序:
public class JRS310Serializer extends JsonSerializer<MyObject> {
public JRS310Serializer() {
super();
objectMapper.registerModule(new JavaTimeModule());
objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
}
}
我把它添加到应用程序。
spring:
kafka:
producer:
value-serializer: org.package.JRS310Serializer
我还把这个添加到application.yaml:
spring:
jackson:
serialization:
WRITE_DATES_AS_TIMESTAMPS: false
但是没有任何帮助。在调试过程中,我看到我的自定义序列化器中的objectMapper创建正确。如果我使用以下命令将序列化器提供的objectMapper示例调用到KafkaProducer,我甚至可以获得正常的json:
objectMapper.writeValueAsString(MyObject.builder().eventTime(LocalDateTime.now()).build());
但问题是,在序列化JSON时,Kafka使用ObjectWriter类示例(而不是ObjectMapper)和以下命令
this.writer.writeValueAsBytes(MyObject.builder().eventTime(LocalDateTime.now()).build())
而且它总是返回错误的数组值。首先我在2.7.7 Sping Boot 中尝试了托管依赖项。然后更新到2.7.8,但没有帮助。这是一个新的bug还是我做错了什么?提前感谢。
1条答案
按热度按时间unhi4e5o1#
帮助重写公共byte[] serialize(字符串主题,@Nullable T数据)JsonSerializer方法,其中我使用ObjectMapper而不是ObjectWriter: