Spring Kafka Класс не входит в доверенные пакеты

См. документация.

Starting with version 2.1, type information can be conveyed in record Headers, allowing the handling of multiple types. In addition, the serializer/deserializer can be configured using Kafka properties.

JsonSerializer.ADD_TYPE_INFO_HEADERS (default true); set to false to disable this feature on the JsonSerializer (sets the addTypeInfo property).

JsonDeserializer.KEY_DEFAULT_TYPE; fallback type for deserialization of keys if no header information is present.

JsonDeserializer.VALUE_DEFAULT_TYPE; fallback type for deserialization of values if no header information is present.

JsonDeserializer.TRUSTED_PACKAGES (default java.util, java.lang); comma-delimited list of package patterns allowed for deserialization; * means deserialize all.

По умолчанию сериализатор добавляет в заголовки информацию о типе.

см. документация по загрузке.

Similarly, you can disable the JsonSerializer default behavior of sending type information in headers:

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializerspring.kafka.producer.properties.spring.json.add.type.headers=false

Или вы можете добавить сопоставление типов в конвертер входящих сообщений, чтобы сопоставить тип источника с типом назначения.

РЕДАКТИРОВАТЬ

Сказав это, какую версию вы используете?

Следует отметить два ключевых момента.

1. Есть два отдельных проекта для продюсера и потребителя.
2. Тогда отправка сообщения (значения) - это довольно примитивный тип объекта.

Проблема в том, что объект создания сообщения недоступен на стороне потребителя, потому что это два отдельных проекта.

Два решения этой проблемы, пожалуйста, следуйте нижеуказанным шагам в приложениях Spring boot Producer и Consumer.

---- Приложение производителя -------------

** Класс конфигурации производителя **

import com.kafka.producer.models.Container;    
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

@Bean
public ProducerFactory<String, Container> producerFactory(){

    Map<String, Object> config = new HashMap<>();

config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

    return new DefaultKafkaProducerFactory(config);
}

@Bean
public KafkaTemplate<String, Container> kafkaTemplate(){
    return new KafkaTemplate<>(producerFactory());
}
}

Примечание. Контейнер - это настраиваемый объект, который будет размещен в теме kafka.

** Класс производителя **

import com.kafka.producer.models.Container;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class Producer {

private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "final-topic";

@Autowired
private KafkaTemplate<String, Container> kafkaTemplate;

public void sendUserMessage(Container msg) {
    LOGGER.info(String.format("\n ===== Producing message in JSON ===== \n"+msg));
    Message<Container> message = MessageBuilder
            .withPayload(msg)
            .setHeader(KafkaHeaders.TOPIC, TOPIC)
            .build();
    this.kafkaTemplate.send(message);
}
}

** Контроллер производителя **

import com.kafka.producer.models.Container;
import com.kafka.producer.services.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/message")
public class MessageController {

@Autowired
private Producer producer;

@PostMapping(value = "/publish")
public String sendMessageToKafkaTopic(@RequestBody Container containerMsg) {
    this.producer.sendUserMessage(containerMsg);
    return "Successfully Published !!";
}
}

Примечание. Сообщение с типом Контейнер будет опубликовано в теме kafka name: final-topic как сообщение JSON.

================================================== =============================

- Потребительское приложение -

** Класс конфигурации **

import com.kafka.consumer.models.Container;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConsumerOneConfig {

@Bean
public ConsumerFactory<String, Container> consumerFactory(){
    JsonDeserializer<Container> deserializer = new JsonDeserializer<>(Container.class);
    deserializer.setRemoveTypeHeaders(false);
    deserializer.addTrustedPackages("*");
    deserializer.setUseTypeMapperForKey(true);

    Map<String, Object> config = new HashMap<>();

    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_one");
    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer);

    return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), deserializer);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Container> kafkaListenerContainerFactory(){
    ConcurrentKafkaListenerContainerFactory<String, Container> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;

}
}

Примечание. Здесь вы можете видеть, что вместо использования JsonDeserializer () по умолчанию мы должны использовать настраиваемый JsonDeserializer для использования сообщений Json типа объекта контейнера из final-topic (название темы).

** Бытовое обслуживание **

import com.kafka.consumer.models.Container;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import java.io.IOException;

@Service
public class ConsumerOne {

private final Logger LOGGER = LoggerFactory.getLogger(ConsumerOne.class);

@KafkaListener(topics = "final-topic", groupId = "group_one", containerFactory = "kafkaListenerContainerFactory")
public void consumeUserMessage(@Payload Container msg, @Headers MessageHeaders headers) throws IOException {
    System.out.println("received data in Consumer One  = "+ msg.getMessageTypes());
}
}

jsonDeserializer.addTrustedPackages("*");

решил мою проблему для spring-kafka-2.2.8.

Чтобы добавить его в application.properties:

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

ВАЖНАЯ ЗАМЕТКА:

They have no effect if you have provided Serializer and Deserializer instances for KafkaConsumer and KafkaProducer, respectively.

Ссылки:
[1] https://docs.spring.io/spring-kafka/reference/html/#json-serde
[2] https://github.com/spring-projects/spring-kafka/issues/535

Для этого есть два способа сделать это: в десериализаторе или в application.yml.

Либо в вашем десериализаторе

В десериализаторе, который вы используете в своем DefaultKafkaConsumerFactory (для создания своей фабрики потребителей). Допустим, вы хотите создать ConsumerFactory<String, Foo> , где Foo является моделью / POJO, которую вы хотите использовать для своих сообщений kafka.

Вам нужен addTrustedPackages из JsonDeserializer У меня есть пример в Kotlin, но в java почти такой же синтаксис:

 val deserializer = JsonDeserializer<Foo>()
 deserializer.addTrustedPackages("com.example.entity.Foo") // Adding Foo to our trusted packages

 val consumerFactory = DefaultKafkaConsumerFactory(
      consumerConfigs(),  // your consumer config
      StringDeserializer(), 
      deserializer // Using our newly created deserializer
 )

Или в вашем application.yml

В вашем файле application.yml следуйте инструкциям весна-кафка. Мы добавляем класс Foo из пакета com.example.entity.Foo в доверенное хранилище, используя:

spring:
  kafka:
    consumer:
      properties:
        spring.json.trusted.packages: "com.example.entity.Foo"

С spring.json.trusted.packages, принимающим массив пакетов. Вы можете указать пакет класса или использовать * для любых пакетов. В этом случае вам не нужно передавать свой deserializer в DefaultKafkaConsumerFactory() только в конфигурации потребителя.

Я также столкнулся с этой проблемой, однако приведенные выше решения не помогли мне. Однако в чем заключалась хитрость, так это в настройке фабрики потребителей kafka следующим образом:

props.put(JsonDeserializer.TRUSTED_PACKAGES, "your.package.name");

Моя версия spring-kafka - 2.2.11, и у меня тоже была эта ошибка.

Я получил эту ошибку, потому что я настроил двух потребителей в одной теме kafta с разной конфигурацией. У одного из них был ConsumerFactory <String, OrderDTO>, а у другого ConsumerFactory <String, String>.

Я решил ошибку, изменив конфигурацию одного потребителя, потому что был неправ.

Просто проверяете потребителей темы