title: Spring Kafka The class is not in the trusted packages
body: <p>In my Spring Boot/Kafka application before the library update, I used the following class <code>org.telegram.telegrambots.api.objects.Update</code> in order to post messages to the Kafka topic. Right now I use the following <code>org.telegram.telegrambots.meta.api.objects.Update</code>. As you may see - they have different packages.</p>
  
  <p>After application restart I ran into the following issue:</p>
  
  <pre><code>[org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1] o.s.kafka.listener.LoggingErrorHandler : Error while processing: null
  
  org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition telegram.fenix.bot.update-0 at offset 4223. If needed, please seek past the record to continue consumption.
  Caused by: java.lang.IllegalArgumentException: The class 'org.telegram.telegrambots.api.objects.Update' is not in the trusted packages: [java.util, java.lang, org.telegram.telegrambots.meta.api.objects]. 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 (*).
  at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:139) ~[spring-kafka-2.1.8.RELEASE.jar!/:2.1.8.RELEASE]
  at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:113) ~[spring-kafka-2.1.8.RELEASE.jar!/:2.1.8.RELEASE]
  at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:221) ~[spring-kafka-2.1.8.RELEASE.jar!/:2.1.8.RELEASE]
  at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:967) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.internals.Fetcher.access$3300(Fetcher.java:93) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1144) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1400(Fetcher.java:993) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:527) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:488) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1155) ~[kafka-clients-1.1.0.jar!/:na]
  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1115) ~[kafka-clients-1.1.0.jar!/:na]
  at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:699) ~[spring-kafka-2.1.8.RELEASE.jar!/:2.1.8.RELEASE]
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_171]
  at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_171]
  at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
  </code></pre>
  
  <p>This is my config:</p>
  
  <pre><code>@EnableAsync
  @Configuration
  public class ApplicationConfig {
  
      @Bean
      public StringJsonMessageConverter jsonConverter() {
          return new StringJsonMessageConverter();
      }
  
  }
  
  @Configuration
  public class KafkaProducerConfig {
  
      @Value("${spring.kafka.bootstrap-servers}")
      private String bootstrapServers;
  
      @Bean
      public Map&lt;String, Object&gt; producerConfigs() {
  
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
  
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
          props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 15000000);
  
          return props;
      }
  
      @Bean
      public ProducerFactory&lt;String, Update&gt; updateProducerFactory() {
          return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
      }
  
      @Bean
      public KafkaTemplate&lt;String, Update&gt; updateKafkaTemplate() {
          return new KafkaTemplate&lt;&gt;(updateProducerFactory());
      }
  
  }
  
  @Configuration
  public class KafkaConsumerConfig {
  
      @Value("${kafka.consumer.max.poll.interval.ms}")
      private String kafkaConsumerMaxPollIntervalMs;
  
      @Value("${kafka.consumer.max.poll.records}")
      private String kafkaConsumerMaxPollRecords;
  
      @Value("${kafka.topic.telegram.fenix.bot.update.consumer.concurrency}")
      private Integer updateConsumerConcurrency;
  
      @Bean
      public ConsumerFactory&lt;String, String&gt; consumerFactory(KafkaProperties kafkaProperties) {
          return new DefaultKafkaConsumerFactory&lt;&gt;(kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new JsonDeserializer&lt;&gt;(String.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
  
          kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
          kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);
  
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
          factory.setConsumerFactory(consumerFactory(kafkaProperties));
  
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;String, Update&gt; updateConsumerFactory(KafkaProperties kafkaProperties) {
          return new DefaultKafkaConsumerFactory&lt;&gt;(kafkaProperties.buildConsumerProperties(), new StringDeserializer(), new JsonDeserializer&lt;&gt;(Update.class));
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, Update&gt; updateKafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
  
          kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerMaxPollIntervalMs);
          kafkaProperties.getProperties().put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerMaxPollRecords);
  
          ConcurrentKafkaListenerContainerFactory&lt;String, Update&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
          factory.setConsumerFactory(updateConsumerFactory(kafkaProperties));
          factory.setConcurrency(updateConsumerConcurrency);
  
          return factory;
      }
  
  }
  </code></pre>
  
  <p>application.properties</p>
  
  <pre><code>spring.kafka.bootstrap-servers=${kafka.host}:${kafka.port}
  spring.kafka.consumer.auto-offset-reset=earliest
  spring.kafka.consumer.group-id=postfenix
  spring.kafka.consumer.enable-auto-commit=false
  spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
  </code></pre>
  
  <p>How to solve this issue and let Kafka deserialize old messages into the new ones ?</p>
  
  <p><strong>UPDATED</strong></p>
  
  <p>This is my listener</p>
  
  <pre><code>@Component
  public class UpdateConsumer {
  
      @KafkaListener(topics = "${kafka.topic.update}", containerFactory = "updateKafkaListenerContainerFactory")
      public void onUpdateReceived(ConsumerRecord&lt;String, Update&gt; consumerRecord, Acknowledgment ack) {
  
          //do some logic here
  
          ack.acknowledge();
      }
  
  }
  </code></pre>