Use bean CL for JdbcMessageStore.deserializer

Related to https://stackoverflow.com/questions/72305387/spring-integration-delayer-starts-sending-null-message-payloads-when-switched

In some async use-cases (e.g. `DelayHandler`), the context classloader
might be different for the data to be deserialized from message store.

* Fix `JdbcMessageStore` to populate a bean `ClassLoader` into default
`AllowListDeserializingConverter` from the application context.
The provided `Deserializer` must ensure such a `ClassLoader` itself
* Add warning message to the `LambdaMessageProcessor` when converter
returns `null` for the payload it cannot convert to expected type.
Cannot be raised as error since some applications may already rely
on the `null` conversion result in their method arguments

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

spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java

@@ -29,6 +29,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.core.MethodIntrospector;
+import org.springframework.core.log.LogMessage;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
@@ -151,7 +152,14 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
 						args[i] = message;
 					}
 					else {
-						args[i] = this.messageConverter.fromMessage(message, this.expectedType);
+						Object payload = this.messageConverter.fromMessage(message, this.expectedType);
+						if (payload == null && LOGGER.isWarnEnabled()) {
+							LOGGER.warn(LogMessage.format(
+									"The '%s' returned 'null' for the payload conversion from the " +
+											"'%s' and expected type '%s'.",
+									this.messageConverter, message, this.expectedType));
+						}
+						args[i] = payload;
 					}
 				}
 				else {

spring-integration-core/src/main/java/org/springframework/integration/support/converter/AllowListDeserializingConverter.java

@@ -41,6 +41,9 @@ import org.springframework.util.PatternMatchUtils;
  * classes/packages are deserialized. If you receive data from untrusted sources, consider
  * adding trusted classes/packages using {@link #setAllowedPatterns(String...)} or
  * {@link #addAllowedPatterns(String...)}.
+ * <p>
+ * If a delegate deserializer is a {@link DefaultDeserializer}, only its {@link ClassLoader}
+ * is used for a {@link ConfigurableObjectInputStream} logic.
  *
  * @author Gary Russell
  * @author Mark Fisher
@@ -133,7 +136,13 @@ public class AllowListDeserializingConverter implements Converter<byte[], Object
 				return deserialize(byteStream);
 			}
 			else {
-				return this.deserializer.deserialize(byteStream);
+				Object result = this.deserializer.deserialize(byteStream);
+				/* Even if there is no knowledge what is the target deserialization algorithm
+				 and malicious code may be executed already, it is still better to fail
+				 with untrusted data rather than just let it pass downstream.
+				 */
+				checkAllowList(result.getClass());
+				return result;
 			}
 		}
 		catch (Exception ex) {

spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-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.
@@ -19,6 +19,7 @@ package org.springframework.integration.dsl;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
+import java.util.Date;
 import java.util.Objects;
 import java.util.function.Function;
 
@@ -31,6 +32,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.IntegrationConverter;
+import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.handler.GenericHandler;
 import org.springframework.integration.handler.LambdaMessageProcessor;
 import org.springframework.integration.transformer.GenericTransformer;
@@ -85,6 +87,24 @@ public class LambdaMessageProcessorTests {
 				.isThrownBy(() -> lmp.processMessage(testMessage));
 	}
 
+	@Test
+	public void testConversionToNull() {
+		LambdaMessageProcessor lmp = new LambdaMessageProcessor(
+				new GenericSelector<Date>() { // Must not be lambda
+
+					@Override
+					public boolean accept(Date payload) {
+						return payload == null;
+					}
+
+				}, Date.class);
+		lmp.setBeanFactory(this.beanFactory);
+		GenericMessage<String> testMessage = new GenericMessage<>("foo");
+		Object result = lmp.processMessage(testMessage);
+		assertThat(result).isEqualTo(Boolean.TRUE);
+	}
+
+
 	@Test
 	public void testCustomConverter() {
 		LambdaMessageProcessor lmp = new LambdaMessageProcessor(Function.identity(), TestPojo.class);

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java

@@ -31,6 +31,7 @@ import java.util.stream.Stream;
 
 import javax.sql.DataSource;
 
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.core.serializer.support.SerializingConverter;
@@ -79,7 +80,7 @@ import org.springframework.util.StringUtils;
  *
  * @since 2.0
  */
-public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore {
+public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore, BeanClassLoaderAware {
 
 	/**
 	 * Default value for the table prefix property.
@@ -177,7 +178,10 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 	private String tablePrefix = DEFAULT_TABLE_PREFIX;
 
-	private AllowListDeserializingConverter deserializer;
+	private AllowListDeserializingConverter deserializer =
+			new AllowListDeserializingConverter(JdbcMessageStore.class.getClassLoader());
+
+	private boolean deserializerExplicitlySet;
 
 	private SerializingConverter serializer;
 
@@ -199,7 +203,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	public JdbcMessageStore(JdbcOperations jdbcOperations) {
 		Assert.notNull(jdbcOperations, "'dataSource' must not be null");
 		this.jdbcTemplate = jdbcOperations;
-		this.deserializer = new AllowListDeserializingConverter();
 		this.serializer = new SerializingConverter();
 		try {
 			this.vendorName =
@@ -211,6 +214,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 		}
 	}
 
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		if (!this.deserializerExplicitlySet) {
+			this.deserializer = new AllowListDeserializingConverter(classLoader);
+		}
+	}
+
 	/**
 	 * Public setter for the table prefix property. This will be prefixed to all the table names before queries are
 	 * executed. Defaults to {@link #DEFAULT_TABLE_PREFIX}.
@@ -249,12 +259,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	}
 
 	/**
-	 * A converter for deserializing byte arrays to messages.
+	 * A converter for deserializing byte arrays to message.
 	 * @param deserializer the deserializer to set
 	 */
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
 		this.deserializer = new AllowListDeserializingConverter((Deserializer) deserializer);
+		this.deserializerExplicitlySet = true;
 	}
 
 	/**

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 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.
@@ -35,6 +35,7 @@ import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.condition.LongRunningTest;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
@@ -80,6 +81,10 @@ public class DelayerHandlerRescheduleIntegrationTests {
 		MessageChannel input = context.getBean("input", MessageChannel.class);
 		MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
 
+		ClassLoader messageStoreDeserializerClassLoader =
+				TestUtils.getPropertyValue(messageStore, "deserializer.defaultDeserializerClassLoader",
+						ClassLoader.class);
+		assertThat(messageStoreDeserializerClassLoader).isSameAs(context.getClassLoader());
 		assertThat(messageStore.getMessageGroupCount()).isEqualTo(0);
 		Message<String> message1 = MessageBuilder.withPayload("test1").build();
 		input.send(message1);