From 3465110f0703f6e10d68ff488f935594d72ace6c 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 # Conflicts: # spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java --- .../http/multipart/MultipartFileReader.java | 3 +- .../inbound/RedisQueueInboundGateway.java | 40 +++++++++++------- .../RedisQueueMessageDrivenEndpoint.java | 42 ++++++++++--------- .../outbound/RedisQueueOutboundGateway.java | 18 +++++--- .../redis/store/RedisChannelMessageStore.java | 19 +++++++-- .../RedisQueueMessageDrivenEndpointTests.java | 22 +++++----- 6 files changed, 87 insertions(+), 57 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 e3f2abb827..b79950ca2a 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; /** @@ -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}. @@ -157,11 +165,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()); @@ -179,8 +187,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()); @@ -189,7 +197,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 a46f486a1c..bef3935249 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 @@ -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) { @@ -153,12 +161,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; } @@ -186,7 +188,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 d63c94b135..a623e46c9e 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 @@ -43,19 +43,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"); @@ -68,6 +68,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..5d4238c16e 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,17 +42,21 @@ 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. * @@ -65,6 +70,13 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw 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 0504af463b..7c034ba915 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 @@ -68,8 +68,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 @@ -78,8 +78,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 { @@ -99,7 +98,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3014Default() throws Exception { - String queueName = "si.test.redisQueueInboundChannelAdapterTests"; RedisTemplate redisTemplate = new RedisTemplate(); @@ -122,6 +120,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(); @@ -142,8 +141,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); @@ -165,6 +163,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); @@ -193,7 +192,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Test @RedisAvailable public void testInt3017IntegrationInbound() throws Exception { - String payload = new Date().toString(); RedisTemplate redisTemplate = new StringRedisTemplate(); @@ -227,7 +225,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); @@ -262,7 +260,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.boundListOps(queueName).leftPush("foo"); - final CountDownLatch stopLatch = new CountDownLatch(1); + CountDownLatch stopLatch = new CountDownLatch(1); endpoint.stop(new Runnable() { @@ -331,7 +329,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()); @@ -353,7 +351,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3932ReadFromLeft() throws Exception { - String queueName = "si.test.redisQueueInboundChannelAdapterTests3932"; RedisTemplate redisTemplate = new RedisTemplate(); @@ -376,6 +373,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);