spring - kafka中的反序列化异常
问题描述
使用 kafka-streams 时出现以下错误。
[Kafka Stream] 10:09:26.442 ERROR --- o.a.k.s.e.LogAndFailExceptionHandler: Exception caught
during Deserialization, taskId: 0_0, topic: t.commodity.promotion, partition: 0, offset: 0
java.lang.IllegalArgumentException: The class'com.course.kafka.kafkaorder.Broker.Message.PromotionMessage'
is not in the trusted packages: [java.util, java.lang, com.course.stream.broker.message,
com.course.stream.broker.message.*]. If you believe this class is safe to deserialize,
please provide its name. If the serialization is only done by a trusted source, you can also enable
trust all (*).
[Kafka Stream] 10:09:26.444 ERROR --- o.a.k.s.p.internals.StreamThread: stream-thread [kafka-stream-7972
450a-443b-8b7b-007e9fdf8e4c-StreamThread-1] Encountered the following exception during
processing and the thread is going to shut down:
org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to
fail upon a deserialization error. If you would rather have the streaming pipeline continue after a
deserialization error, please set the default.deserialization.exception.handler appropriately.
我的代码是
@Configuration
public class PromotionJsonSerde {
@Bean
public KStream<String, PromotionMessage> kStreamPromotionUppercase(StreamsBuilder builder)
{
var stringSerde = Serdes.String();
var jsonSerde = new JsonSerde<>(PromotionMessage.class);
KStream<String, PromotionMessage> sourceStream = builder.stream("t.commodity.promotion",
Consumed.with(stringSerde, jsonSerde));
KStream<String, PromotionMessage> uppercaseStream = sourceStream.mapValues(this::uppercasePromotionCode);
uppercaseStream.to("t.commodity.promotion-uppercase", Produced.with(stringSerde, jsonSerde));
return sourceStream;
}
private PromotionMessage uppercasePromotionCode(PromotionMessage message)
{
return new PromotionMessage(message.getPromotionCode().toUpperCase());
}
}
促销代码
public class PromotionMessage {
private String promotionCode;
// getters,setters, tostring
}
应用程序.yml
logging:
pattern:
console: "[Kafka Stream] %clr(%d{HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:%5p}) %clr(---){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:%wEx}"
spring:
main:
banner-mode: OFF
kafka:
listener:
missing-topics-fatal: false
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
通过看到错误,我尝试在 application.yml 文件中添加以下代码。但我仍然遇到
同样的错误。
spring:
kafka:
producer:
streams:
properties:
default.deserialization.exception.handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler,org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler
consumer:
properties:
spring:
json:
trusted:
packages: "*"
cloud:
stream:
kafka:
streams:
bindings:
process-in-0.consumer:
deserializationExceptionHandler: logAndContinue
为什么我收到反序列化错误?
如果我检查终端上的消费者,我会收到生产者发送的消息。
谁能帮我 ?
解决方案
我看到您正在自己创建 serde new JsonSerde<>(PromotionMessage.class);
- 我们自动将该类的包添加到受信任的包中;因此
trusted packages: [java.util, java.lang, com.course.stream.broker.message, com.course.stream.broker.message.*]
当您创建自己的 serde 时,该属性将被忽略。反序列化器试图创建com.course.kafka.kafkaorder.Broker.Message.PromotionMessage
在不同包中的;很可能是生产者的不同类别。
添加这个:((JsonDeserializer) jsonSerde.deserializer()).setUseTypeHeaders(false);
告诉反序列化器忽略标头中的类型信息并改用提供的回退类型。
推荐阅读
- java - Camel 总是将消息路由到预期 Kafka 主题的 0 分区
- google-sheets - 在 Google 表格中的单元格中的 URL 中插入动态值
- css - 如何使用 CSS 简单地更改 wordpress 中字段标签的文本?
- windows - 需要帮助从 Windows 10 .BAT 文件运行嵌套程序调用而没有“不支持 UNC 路径”错误
- java - 如何使用多个填充段格式化 Java 字符串
- discord.js - discord.js v12 | TypeError:无法读取未定义的属性“发送”
- mongodb - 如何在 mongo 中使用 $addFields 将元素添加到现有文档?
- java - Spring Data JPA 附加 EntityManagerFactory 仅针对缓存和批量操作进行了优化
- python - 我有一个使用 OpenCV 的 python 脚本,代码在 python 2.7 中有效,但在 python 3.7 中无效,我不知道为什么
- c# - RDLC 报告表达式