Message history not getting deserialized

user_3538074 0
Viewed 112

I was trying to send message history via kafka backed message channel, and i am getting an error like below:

Caused by: java.lang.IllegalArgumentException: Incorrect type specified for header 'history'. Expected [class org.springframework.integration.history.MessageHistory] but actual type is [class org.springframework.kafka.support.DefaultKafkaHeaderMapper$NonTrustedHeaderType]
at org.springframework.messaging.MessageHeaders.get(MessageHeaders.java:216)
at org.springframework.integration.history.MessageHistory.write(MessageHistory.java:96)

Environment:

  • Java version: JDK8
  • Kafka version: 3.1.0
  • Spring-boot-starter-integration: 2.6.2 (integration core:5.5.7)

The message is getting deserialized properly without message history, but unable to do so with message history.

Here is the configuration that I am setting:

Consumer:

  public ConsumerFactory consumerFactory(String groupId, String clientId) {
    Properties consumerProperties = new Properties();
    consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AUTO_OFFSET_RESET_CONFIG);
    consumerProperties
        .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
    consumerProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,MAX_POLL_INTERVAL_MS_CONFIG);
    consumerProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,MAX_POLL_RECORDS_CONFIG);

    DefaultKafkaConsumerFactory defaultKafkaConsumerFactory = new DefaultKafkaConsumerFactory(
        consumerProperties);

    JsonDeserializer jsonDeserializer = new JsonDeserializer(GenericMessage.class, JacksonJsonUtils.messagingAwareMapper());

    jsonDeserializer.addTrustedPackages("*");
    defaultKafkaConsumerFactory.setValueDeserializer(jsonDeserializer);
    return defaultKafkaConsumerFactory;
  }

Producer:

  public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> producerConfigMap = new HashMap<>();
    producerConfigMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    producerConfigMap.put(ProducerConfig.LINGER_MS_CONFIG, PRODUCER_LINGER_MS_CONFIG);
    producerConfigMap.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, PRODUCER_COMPRESSION_TYPE_CONFIG);
    producerConfigMap.put(ProducerConfig.BATCH_SIZE_CONFIG, PRODUCER_BATCH_SIZE_CONFIG);
    producerConfigMap
        .put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    JsonSerializer<GenericMessage> jsonSerializer = new JsonSerializer(JacksonJsonUtils.messagingAwareMapper());

    DefaultKafkaProducerFactory defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(
        producerConfigMap);

    defaultKafkaProducerFactory.setValueSerializer(jsonSerializer);
    return defaultKafkaProducerFactory;
  }

ConcurentKafkaContainerListenerFactory:

  public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
      String groupId, String clientId) {

    DefaultKafkaHeaderMapper kafkaHeaderMapper = new DefaultKafkaHeaderMapper();
    kafkaHeaderMapper.addTrustedPackages("org.springframework.integration.history");

    MessagingMessageConverter messagingMessageConverter = new MessagingMessageConverter();
    messagingMessageConverter.setHeaderMapper(kafkaHeaderMapper);

    ConcurrentKafkaListenerContainerFactory<String, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();

    factory.setConsumerFactory(consumerFactory(groupId, clientId));
    factory.setMessageConverter(messagingMessageConverter);
    return factory;
  }

I have tried adding trusted packages in every possible way, but still, I am getting the above error.

spring-integration
1 Answers

Looks at the exception again: org.springframework.kafka.support.DefaultKafkaHeaderMapper$NonTrustedHeaderType. It doesn't say that something is wrong with your (de)serializers. It is a DefaultKafkaHeaderMapper mapper feature to ban that type from you.

You need to supply a DefaultKafkaHeaderMapper with the addTrustedPackages("*") on the consumer side. If you use KafkaMessageDrivenChannelAdapter, see its setMessageConverter(MessageConverter messageConverter) to be populated with the MessagingMessageConverter. And that one has an option for the setHeaderMapper(KafkaHeaderMapper headerMapper), where you already can set that DefaultKafkaHeaderMapper.

Please, raise a GH, so we can add that org.springframework.integration.history to trusted packages for a default DefaultKafkaHeaderMapper in the KafkaMessageDrivenChannelAdapter.

user_27565470
Related