From 9f4f91d5f9b160d6d61c7f6cbfd826cea8b31db3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 20 May 2022 11:37:43 -0400 Subject: [PATCH] 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`** --- .../handler/LambdaMessageProcessor.java | 10 ++++++++- .../AllowListDeserializingConverter.java | 11 +++++++++- .../dsl/LambdaMessageProcessorTests.java | 22 ++++++++++++++++++- .../jdbc/store/JdbcMessageStore.java | 19 ++++++++++++---- ...ayerHandlerRescheduleIntegrationTests.java | 7 +++++- 5 files changed, 61 insertions(+), 8 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java index 2f77fa23cb..8fa40651fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java +++ b/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, 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 { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/AllowListDeserializingConverter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/converter/AllowListDeserializingConverter.java index 891b79d427..0963712f00 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/converter/AllowListDeserializingConverter.java +++ b/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...)}. + *

+ * 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 lmp.processMessage(testMessage)); } + @Test + public void testConversionToNull() { + LambdaMessageProcessor lmp = new LambdaMessageProcessor( + new GenericSelector() { // Must not be lambda + + @Override + public boolean accept(Date payload) { + return payload == null; + } + + }, Date.class); + lmp.setBeanFactory(this.beanFactory); + GenericMessage 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); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java index 3aafce78f0..8276c02ef0 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcMessageStore.java +++ b/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> deserializer) { this.deserializer = new AllowListDeserializingConverter((Deserializer) deserializer); + this.deserializerExplicitlySet = true; } /** diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java index 71472530d2..1f2f7f0637 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests.java +++ b/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 message1 = MessageBuilder.withPayload("test1").build(); input.send(message1);