SpringKafka模型不在可信包中

a0x5cqrl  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(366)

我在做微服务 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 (*).

所以从错误信息来看这是 com.service1.model.TopicMessage service1的序列化模型。但我正在尝试将消息反序列化到模型中 com.service2.model.ConsumeMessage 它在服务中2并且有这个问题
我在这里发现了同样的问题,并尝试了下面的格式和文档
下面是我的配置

@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);

     }

Kafka消费工厂

@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);

}

KafkalistenerContainer工厂

@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;
}
xkftehaa

xkftehaa1#

你需要禁用 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

相关问题