本文介紹了Spring Kafka模型不在受信任的包中的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！
問題描述
我正在使用spring-Kafka-2.1.5和spring-boot-2.0.5處理微服務
第一個服務將向Kafka生成一些消息，第二個服務將消費這些消息，而消費時我遇到了問題
Caused by: java.lang.IllegalArgumentException: The class 'com.service1.model.TopicMessage' is not in the trusted packages: [java.util, java.lang, com.service2.model.ConsumeMessage].
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 (*).
因此，從錯誤消息來看，這是Service1的com.service1.model.TopicMessage序列化模型。但我正在嘗試將消息反序列化到Service2中的模型com.service2.model.ConsumeMessage中，并出現此問題
我發現了同樣的問題here，并嘗試了下面的格式和文檔docs
下面是我的配置
@Bean(name = "kafkaConsumerConfig")
public Map<String, Object> kafkaConsumerConfig() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetconfig);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
}
kafkaConsumer erFactory
@Bean(name = "kafkaConsumerFactory")
public ConsumerFactory<String, ConsumeMessage> kafkaConsumerFactory() {
JsonDeserializer<ConsumeMessage>
deserializer = new JsonDeserializer<>();
deserializer.addTrustedPackages("com.service2.model");
return new DefaultKafkaConsumerFactory<String, ConsumeMessage>(kafkaConsumerConfig(),new StringDeserializer(),deserializer);
}
kafkaListenerContainerFactory
@Bean(name = "kafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, ConsumeMessage > kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, ConsumeMessage > factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConcurrency(Integer.parseInt(threads));
factory.setBatchListener(true);
factory.setConsumerFactory(kafkaConsumerFactory());
factory.getContainerProperties().setPollTimeout(Long.parseLong(pollTimeout));
factory.getContainerProperties().setAckMode(AckMode.BATCH);
return factory;
}
推薦答案
您需要禁用反序列化程序中的header precedence：
/**
* Construct an instance with the provided target type, and
* useHeadersIfPresent with a default {@link ObjectMapper}.
* @param targetType the target type.
* @param useHeadersIfPresent true to use headers if present and fall back to target
* type if not.
* @since 2.2
*/
public JsonDeserializer(Class<? super T> targetType, boolean useHeadersIfPresent) {
必須將useHeadersIfPresent參數配置為false。這樣，將使用inferred類型，并且將忽略標頭值。
如果您不使用spring-kafka-2.2，則應考慮使用類似的邏輯實現您自己的JsonDeserializer：https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java
這篇關于Spring Kafka模型不在受信任的包中的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
