From eb31825bad513dc02f15dfbb0226a238f738f59d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 28 Jun 2019 15:51:59 -0400 Subject: [PATCH] Fix Redis components for JDK deserialization It turns out that `JdkSerializationRedisSerializer` by default is based on the default Java class loader which may lead into `ClassCastException` downstream after deserialization * Make all the `JdkSerializationRedisSerializer` usage (default) in Redis components based on the BeanFactory `ClassLoader` * Fix tests to call `setBeanClassLoader()` * Fix Mark Fisher's name in the `MultipartFileReader` :-) **Cherry-pick to 5.1.x, 5.0.x & 4.3.x after restoring diamonds** # Conflicts: # spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java # spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java # Conflicts: # spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java # spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java --- .../http/multipart/MultipartFileReader.java | 3 +- .../inbound/RedisQueueInboundGateway.java | 42 +++++++++++------- .../RedisQueueMessageDrivenEndpoint.java | 44 ++++++++++--------- .../outbound/RedisQueueOutboundGateway.java | 18 +++++--- .../redis/store/RedisChannelMessageStore.java | 21 +++++++-- .../RedisQueueMessageDrivenEndpointTests.java | 24 +++++----- 6 files changed, 91 insertions(+), 61 deletions(-) diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/multipart/MultipartFileReader.java b/spring-integration-http/src/main/java/org/springframework/integration/http/multipart/MultipartFileReader.java index 09c8445991..4144c98841 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/multipart/MultipartFileReader.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/multipart/MultipartFileReader.java @@ -23,7 +23,8 @@ import org.springframework.web.multipart.MultipartFile; /** * Strategy for reading {@link MultipartFile} content. * - * @author mark Fisher + * @author Mark Fisher + * * @since 2.0 */ public interface MultipartFileReader { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index fa1ca96196..4d61c6fe7f 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java @@ -19,6 +19,7 @@ package org.springframework.integration.redis.inbound; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -52,7 +53,8 @@ import org.springframework.util.Assert; */ @ManagedResource @IntegrationManagedResource -public class RedisQueueInboundGateway extends MessagingGatewaySupport implements ApplicationEventPublisherAware { +public class RedisQueueInboundGateway extends MessagingGatewaySupport + implements ApplicationEventPublisherAware, BeanClassLoaderAware { private static final String QUEUE_NAME_SUFFIX = ".reply"; @@ -66,24 +68,24 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements private final BoundListOperations boundListOperations; - private volatile ApplicationEventPublisher applicationEventPublisher; + private ApplicationEventPublisher applicationEventPublisher; - private volatile boolean serializerExplicitlySet; + private boolean serializerExplicitlySet; - private volatile Executor taskExecutor; + private Executor taskExecutor; - private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + private RedisSerializer serializer; - private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + + private boolean extractPayload = true; private volatile boolean active; private volatile boolean listening; - private volatile boolean extractPayload = true; - private volatile Runnable stopCallback; /** @@ -93,7 +95,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements public RedisQueueInboundGateway(String queueName, RedisConnectionFactory connectionFactory) { Assert.hasText(queueName, "'queueName' is required"); Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); - this.template = new RedisTemplate(); + this.template = new RedisTemplate<>(); this.template.setConnectionFactory(connectionFactory); this.template.setEnableDefaultSerializer(false); this.template.setKeySerializer(new StringRedisSerializer()); @@ -110,12 +112,18 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements this.applicationEventPublisher = applicationEventPublisher; } + @Override + public void setBeanClassLoader(ClassLoader beanClassLoader) { + if (!this.serializerExplicitlySet) { + this.serializer = new JdkSerializationRedisSerializer(beanClassLoader); + } + } + public void setSerializer(RedisSerializer serializer) { this.serializer = serializer; this.serializerExplicitlySet = true; } - /** * This timeout (milliseconds) is used when retrieving elements from the queue * specified by {@link #boundListOperations}. @@ -147,11 +155,11 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements Assert.notNull(this.serializer, "'serializer' has to be provided where 'extractPayload == false'."); } if (this.taskExecutor == null) { - String beanName = this.getComponentName(); + String beanName = getComponentName(); this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") - + this.getComponentType()); + + getComponentType()); } - if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && this.getBeanFactory() != null) { + if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && getBeanFactory() != null) { MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory())); errorHandler.setDefaultErrorChannel(getErrorChannel()); @@ -169,8 +177,8 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements if (this.active) { logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e); - this.publishException(e); - this.sleepBeforeRecoveryAttempt(); + publishException(e); + sleepBeforeRecoveryAttempt(); } else { logger.debug("Failed to execute listening task. " + e.getClass() + ": " + e.getMessage()); @@ -179,7 +187,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements @SuppressWarnings("unchecked") private void receiveAndReply() { - byte[] value = null; + byte[] value; try { value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index 00397b8bfc..d259e1e730 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -19,6 +19,7 @@ package org.springframework.integration.redis.inbound; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -38,7 +39,6 @@ import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; @@ -55,7 +55,8 @@ import org.springframework.util.Assert; */ @ManagedResource @IntegrationManagedResource -public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware { +public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport + implements ApplicationEventPublisherAware, BeanClassLoaderAware { public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; @@ -63,19 +64,21 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl private final BoundListOperations boundListOperations; - private volatile ApplicationEventPublisher applicationEventPublisher; + private ApplicationEventPublisher applicationEventPublisher; - private volatile MessageChannel errorChannel; + private Executor taskExecutor; - private volatile Executor taskExecutor; + private RedisSerializer serializer; - private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + private boolean serializerExplicitlySet; - private volatile boolean expectMessage = false; + private boolean expectMessage = false; - private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + + private boolean rightPop = true; private volatile boolean active; @@ -83,8 +86,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl private volatile Runnable stopCallback; - private volatile boolean rightPop = true; - /** * @param queueName Must not be an empty String * @param connectionFactory Must not be null @@ -92,7 +93,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl public RedisQueueMessageDrivenEndpoint(String queueName, RedisConnectionFactory connectionFactory) { Assert.hasText(queueName, "'queueName' is required"); Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); - RedisTemplate template = new RedisTemplate(); + RedisTemplate template = new RedisTemplate<>(); template.setConnectionFactory(connectionFactory); template.setEnableDefaultSerializer(false); template.setKeySerializer(new StringRedisSerializer()); @@ -105,8 +106,16 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl this.applicationEventPublisher = applicationEventPublisher; } + @Override + public void setBeanClassLoader(ClassLoader beanClassLoader) { + if (!this.serializerExplicitlySet) { + this.serializer = new JdkSerializationRedisSerializer(beanClassLoader); + } + } + public void setSerializer(RedisSerializer serializer) { this.serializer = serializer; + this.serializerExplicitlySet = true; } /** @@ -114,8 +123,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl * just the payload for a Message, or does the data represent a serialized * {@link Message}?. {@code expectMessage} defaults to false. This means * the retrieved data will be used as the payload for a new Spring Integration - * Message. Otherwise, the data is deserialized as Spring Integration - * Message. + * Message. Otherwise, the data is deserialized as Spring Integration Message. * @param expectMessage Defaults to false */ public void setExpectMessage(boolean expectMessage) { @@ -142,12 +150,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl this.taskExecutor = taskExecutor; } - @Override - public void setErrorChannel(MessageChannel errorChannel) { - super.setErrorChannel(errorChannel); - this.errorChannel = errorChannel; - } - public void setRecoveryInterval(long recoveryInterval) { this.recoveryInterval = recoveryInterval; } @@ -175,7 +177,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && this.getBeanFactory() != null) { MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(this.getBeanFactory())); - errorHandler.setDefaultErrorChannel(this.errorChannel); + errorHandler.setDefaultErrorChannel(getErrorChannel()); this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java index b58a21e61f..6c83b6e32e 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java @@ -44,19 +44,19 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator(); - private final static RedisSerializer stringSerializer = new StringRedisSerializer(); + private static final RedisSerializer stringSerializer = new StringRedisSerializer(); private final RedisTemplate template; private final BoundListOperations boundListOps; - private volatile boolean extractPayload = true; + private boolean extractPayload = true; - private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + private RedisSerializer serializer; - private volatile boolean serializerExplicitlySet; + private boolean serializerExplicitlySet; - private volatile int receiveTimeout = TIMEOUT; + private int receiveTimeout = TIMEOUT; public RedisQueueOutboundGateway(String queueName, RedisConnectionFactory connectionFactory) { Assert.hasText(queueName, "'queueName' is required"); @@ -69,6 +69,14 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand this.boundListOps = this.template.boundListOps(queueName); } + @Override + public void setBeanClassLoader(ClassLoader beanClassLoader) { + super.setBeanClassLoader(beanClassLoader); + if (!this.serializerExplicitlySet) { + this.serializer = new JdkSerializationRedisSerializer(beanClassLoader); + } + } + public void setReceiveTimeout(int timeout) { this.receiveTimeout = timeout; } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java index f6cb1a9379..1ae35c8f79 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java @@ -19,6 +19,7 @@ package org.springframework.integration.redis.store; import java.util.List; import java.util.Set; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.data.redis.connection.RedisConnectionFactory; @@ -41,30 +42,41 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Artem Bilan + * * @since 4.0 * */ -public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAware, InitializingBean { +public class RedisChannelMessageStore + implements ChannelMessageStore, BeanNameAware, InitializingBean, BeanClassLoaderAware { private final RedisTemplate> redisTemplate; - private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); - private String beanName; + private MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); + + private boolean valueSerializerExplicitlySet; + /** * Construct a message store that uses Java Serialization for messages. * * @param connectionFactory The redis connection factory. */ public RedisChannelMessageStore(RedisConnectionFactory connectionFactory) { - this.redisTemplate = new RedisTemplate>(); + this.redisTemplate = new RedisTemplate<>(); this.redisTemplate.setConnectionFactory(connectionFactory); this.redisTemplate.setKeySerializer(new StringRedisSerializer()); this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); this.redisTemplate.afterPropertiesSet(); } + @Override + public void setBeanClassLoader(ClassLoader classLoader) { + if (!this.valueSerializerExplicitlySet) { + this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer(classLoader)); + } + } + /** * Use a different serializer (default {@link JdkSerializationRedisSerializer} for * the {@link Message}. @@ -74,6 +86,7 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw public void setValueSerializer(RedisSerializer valueSerializer) { Assert.notNull(valueSerializer, "'valueSerializer' must not be null"); this.redisTemplate.setValueSerializer(valueSerializer); + this.valueSerializerExplicitlySet = true; } /** diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index b31a296f91..3b0c9dd1b1 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -69,8 +69,8 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.ClassUtils; /** * @author Gunnar Hillert @@ -79,8 +79,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Rainer Frey * @since 3.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) @DirtiesContext public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @@ -100,7 +99,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3014Default() throws Exception { - String queueName = "si.test.redisQueueInboundChannelAdapterTests"; RedisTemplate redisTemplate = new RedisTemplate(); @@ -123,6 +121,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); + endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader()); endpoint.setOutputChannel(channel); endpoint.setReceiveTimeout(10); endpoint.afterPropertiesSet(); @@ -143,8 +142,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3014ExpectMessageTrue() throws Exception { - - final String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; + String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; RedisTemplate redisTemplate = new RedisTemplate(); redisTemplate.setConnectionFactory(this.connectionFactory); @@ -166,6 +164,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); + endpoint.setBeanClassLoader(null); endpoint.setExpectMessage(true); endpoint.setOutputChannel(channel); endpoint.setErrorChannel(errorChannel); @@ -194,7 +193,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Test @RedisAvailable public void testInt3017IntegrationInbound() throws Exception { - String payload = new Date().toString(); RedisTemplate redisTemplate = new StringRedisTemplate(); @@ -228,7 +226,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3442ProperlyStop() throws Exception { - final String queueName = "si.test.testInt3442ProperlyStopTest"; + String queueName = "si.test.testInt3442ProperlyStopTest"; final RedisTemplate redisTemplate = new RedisTemplate(); redisTemplate.setConnectionFactory(this.connectionFactory); @@ -263,9 +261,9 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.boundListOps(queueName).leftPush("foo"); - final CountDownLatch stopLatch = new CountDownLatch(1); + CountDownLatch stopLatch = new CountDownLatch(1); - endpoint.stop(() -> stopLatch.countDown()); + endpoint.stop(stopLatch::countDown); executorService.shutdown(); assertTrue(executorService.awaitTermination(20, TimeUnit.SECONDS)); @@ -314,7 +312,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { ((InitializingBean) this.connectionFactory).afterPropertiesSet(); - RedisTemplate redisTemplate = new RedisTemplate(); + RedisTemplate redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest()); redisTemplate.setEnableDefaultSerializer(false); redisTemplate.setKeySerializer(new StringRedisSerializer()); @@ -336,7 +334,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3932ReadFromLeft() throws Exception { - String queueName = "si.test.redisQueueInboundChannelAdapterTests3932"; RedisTemplate redisTemplate = new RedisTemplate(); @@ -359,6 +356,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); + endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader()); endpoint.setOutputChannel(channel); endpoint.setReceiveTimeout(10); endpoint.setRightPop(false);