GH-3734: Support MessageHistory on Kafka deser

Fixes #3734

The `MessageHistory` is not in the trusted packages of the `DefaultKafkaHeaderMapper`
therefore it fails when `MessageHistory.read()` is performed.

* Configure default `DefaultKafkaHeaderMapper` in the `SubscribableKafkaChannel`,
`KafkaInboundGateway`, `KafkaMessageDrivenChannelAdapter` and `KafkaMessageSource`
to also trust packages exposed via `JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES`
which include a `MessageHistory`, too.

* Verify in some integration Kafka tests to be sure that `MessageHistory` is
deserialized properly in the transferred headers
* Rework some tests to use `@EmbeddedKafka` instead of `@BeforeAll/@AfterAll`

**Cherry-pick to `5.5.x`**

Author: artembilan

Diff for: spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +51,20 @@
  */
 public final class JacksonJsonUtils {
 
+	/**
+	 * The packages to trust on JSON deserialization by default.
+	 */
+	public static final List<String> DEFAULT_TRUSTED_PACKAGES =
+			List.of(
+					"java.util",
+					"java.lang",
+					"org.springframework.messaging.support",
+					"org.springframework.integration.support",
+					"org.springframework.integration.message",
+					"org.springframework.integration.store",
+					"org.springframework.integration.history"
+			);
+
 	private JacksonJsonUtils() {
 	}
 
@@ -99,14 +113,12 @@ public static ObjectMapper messagingAwareMapper(String... trustedPackages) {
 
 	/**
 	 * An implementation of {@link ObjectMapper.DefaultTypeResolverBuilder}
-	 * that wraps a default {@link TypeIdResolver} to the {@link AllowlistTypeIdResolver}.
+	 * that wraps a default {@link TypeIdResolver} to the {@link AllowListTypeIdResolver}.
 	 *
 	 * @author Rob Winch
 	 * @author Artem Bilan
 	 * @author Filip Hanik
 	 * @author Gary Russell
-	 *
-	 * @since 4.3.11
 	 */
 	private static final class AllowListTypeResolverBuilder extends ObjectMapper.DefaultTypeResolverBuilder {
 
@@ -133,8 +145,9 @@ protected TypeIdResolver idResolver(MapperConfig<?> config,
 				JavaType baseType,
 				PolymorphicTypeValidator subtypeValidator,
 				Collection<NamedType> subtypes, boolean forSer, boolean forDeser) {
+
 			TypeIdResolver result = super.idResolver(config, baseType, subtypeValidator, subtypes, forSer, forDeser);
-			return new AllowlistTypeIdResolver(result, this.trustedPackages);
+			return new AllowListTypeIdResolver(result, this.trustedPackages);
 		}
 
 	}
@@ -146,27 +159,14 @@ protected TypeIdResolver idResolver(MapperConfig<?> config,
 	 *
 	 * @author Rob Winch
 	 * @author Artem Bilan
-	 *
-	 * @since 4.3.11
 	 */
-	private static final class AllowlistTypeIdResolver implements TypeIdResolver {
-
-		private static final List<String> TRUSTED_PACKAGES =
-				Arrays.asList(
-						"java.util",
-						"java.lang",
-						"org.springframework.messaging.support",
-						"org.springframework.integration.support",
-						"org.springframework.integration.message",
-						"org.springframework.integration.store",
-						"org.springframework.integration.history"
-				);
+	private static final class AllowListTypeIdResolver implements TypeIdResolver {
 
 		private final TypeIdResolver delegate;
 
-		private final Set<String> trustedPackages = new LinkedHashSet<>(TRUSTED_PACKAGES);
+		private final Set<String> trustedPackages = new LinkedHashSet<>(DEFAULT_TRUSTED_PACKAGES);
 
-		AllowlistTypeIdResolver(TypeIdResolver delegate, String... trustedPackages) {
+		AllowListTypeIdResolver(TypeIdResolver delegate, String... trustedPackages) {
 			this.delegate = delegate;
 			if (trustedPackages != null) {
 				for (String trustedPackage : trustedPackages) {

Diff for: spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,13 +22,18 @@
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
+import org.springframework.integration.support.json.JacksonJsonUtils;
 import org.springframework.integration.support.management.ManageableSmartLifecycle;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.JacksonPresent;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
@@ -49,6 +54,8 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 
 	private final KafkaListenerContainerFactory<?> factory;
 
+	private final IntegrationRecordMessageListener recordListener = new IntegrationRecordMessageListener();
+
 	private MessageDispatcher dispatcher;
 
 	private MessageListenerContainer container;
@@ -71,6 +78,24 @@ public SubscribableKafkaChannel(KafkaOperations<?, ?> template, KafkaListenerCon
 		super(template, channelTopic);
 		Assert.notNull(factory, "'factory' cannot be null");
 		this.factory = factory;
+
+		if (JacksonPresent.isJackson2Present()) {
+			MessagingMessageConverter messageConverter = new MessagingMessageConverter();
+			DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper();
+			headerMapper.addTrustedPackages(JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES.toArray(new String[0]));
+			messageConverter.setHeaderMapper(headerMapper);
+			this.recordListener.setMessageConverter(messageConverter);
+		}
+	}
+
+
+	/**
+	 * Set the {@link RecordMessageConverter} to the listener.
+	 * @param messageConverter the converter.
+	 * @since 6.0
+	 */
+	public void setMessageConverter(RecordMessageConverter messageConverter) {
+		this.recordListener.setMessageConverter(messageConverter);
 	}
 
 	@Override
@@ -113,7 +138,7 @@ protected void onInit() {
 		String groupId = getGroupId();
 		ContainerProperties containerProperties = this.container.getContainerProperties();
 		containerProperties.setGroupId(groupId != null ? groupId : getBeanName());
-		containerProperties.setMessageListener(new IntegrationRecordMessageListener());
+		containerProperties.setMessageListener(this.recordListener);
 	}
 
 	protected MessageDispatcher createDispatcher() {

Diff for: spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java

@@ -35,16 +35,20 @@
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.json.JacksonJsonUtils;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.JacksonPresent;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -106,6 +110,13 @@ public KafkaInboundGateway(AbstractMessageListenerContainer<K, V> messageListene
 		this.messageListenerContainer.setAutoStartup(false);
 		this.kafkaTemplate = kafkaTemplate;
 		setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy());
+		if (JacksonPresent.isJackson2Present()) {
+			MessagingMessageConverter messageConverter = new MessagingMessageConverter();
+			DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper();
+			headerMapper.addTrustedPackages(JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES.toArray(new String[0]));
+			messageConverter.setHeaderMapper(headerMapper);
+			this.listener.setMessageConverter(messageConverter);
+		}
 	}
 
 	/**
@@ -158,6 +169,7 @@ public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
 	 */
 	public void setOnPartitionsAssignedSeekCallback(
 			BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
+
 		this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback;
 	}
 

Diff for: spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java

@@ -17,11 +17,11 @@
 package org.springframework.integration.kafka.inbound;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -36,6 +36,7 @@
 import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.json.JacksonJsonUtils;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.BatchMessageListener;
 import org.springframework.kafka.listener.ConsumerSeekAware;
@@ -47,11 +48,14 @@
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.JacksonPresent;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.BatchMessageConverter;
 import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
 import org.springframework.kafka.support.converter.MessageConverter;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -128,6 +132,15 @@ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer<K, V> m
 		this.messageListenerContainer.setAutoStartup(false);
 		this.mode = mode;
 		setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy());
+
+		if (JacksonPresent.isJackson2Present()) {
+			MessagingMessageConverter messageConverter = new MessagingMessageConverter();
+			DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper();
+			headerMapper.addTrustedPackages(JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES.toArray(new String[0]));
+			messageConverter.setHeaderMapper(headerMapper);
+			this.recordListener.setMessageConverter(messageConverter);
+			this.batchListener.setMessageConverter(messageConverter);
+		}
 	}
 
 	/**
@@ -146,7 +159,6 @@ else if (messageConverter instanceof BatchMessageConverter) {
 			throw new IllegalArgumentException(
 					"Message converter must be a 'RecordMessageConverter' or 'BatchMessageConverter'");
 		}
-
 	}
 
 	/**
@@ -526,9 +538,9 @@ public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowl
 				message = toMessagingMessage(records, acknowledgment, consumer);
 				setAttributesIfNecessary(records, message);
 			}
-			catch (RuntimeException e) {
+			catch (RuntimeException ex) {
 				Exception exception = new ConversionException("Failed to convert to message",
-						records.stream().collect(Collectors.toList()), e);
+						new ArrayList<>(records), ex);
 				MessageChannel errorChannel = getErrorChannel();
 				if (errorChannel != null) {
 					getMessagingTemplate().send(errorChannel, new ErrorMessage(exception));

Diff for: spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -51,13 +51,16 @@
 import org.springframework.integration.core.Pausable;
 import org.springframework.integration.endpoint.AbstractMessageSource;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.support.json.JacksonJsonUtils;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
 import org.springframework.kafka.listener.ConsumerProperties;
 import org.springframework.kafka.listener.ListenerUtils;
 import org.springframework.kafka.listener.LoggingCommitCallback;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.JacksonPresent;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -233,6 +236,12 @@ public KafkaMessageSource(ConsumerFactory<K, V> consumerFactory,
 		this.assignTimeout =
 				Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT)); // NOSONAR - magic
 		this.commitTimeout = consumerProperties.getSyncCommitTimeout();
+
+		if (JacksonPresent.isJackson2Present()) {
+			DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper();
+			headerMapper.addTrustedPackages(JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES.toArray(new String[0]));
+			((MessagingMessageConverter) this.messageConverter).setHeaderMapper(headerMapper);
+		}
 	}
 
 	/**

Diff for: spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java

@@ -29,6 +29,8 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.kafka.channel.PollableKafkaChannel;
 import org.springframework.integration.kafka.channel.PublishSubscribeKafkaChannel;
 import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
@@ -53,6 +55,7 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -70,10 +73,18 @@ void subscribablePtp(@Autowired SubscribableChannel ptp) throws InterruptedExcep
 			latch.countDown();
 		});
 		Message<?> msg = new GenericMessage<>("foo");
+		NullChannel component = new NullChannel();
+		component.setBeanName("myNullChannel");
+		msg = MessageHistory.write(msg, component);
 		ptp.send(msg, 10_000L);
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		assertThat(message.get().getPayload()).isEqualTo("foo");
-		assertThat(message.get().getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.1");
+		Message<?> received = message.get();
+		assertThat(received.getPayload()).isEqualTo("foo");
+		assertThat(received.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.1");
+
+		MessageHistory messageHistory = MessageHistory.read(received);
+		assertThat(messageHistory).isNotNull();
+		assertThat(messageHistory.toString()).isEqualTo("myNullChannel");
 	}
 
 	@Test