From f89e8aafa5edc25f2c331c045df2e91f2e6fedc5 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** --- .../http/multipart/MultipartFileReader.java | 3 +- .../inbound/RedisQueueInboundGateway.java | 42 +++++++++++------- .../RedisQueueMessageDrivenEndpoint.java | 44 ++++++++++--------- .../outbound/RedisQueueOutboundGateway.java | 18 +++++--- .../redis/store/RedisChannelMessageStore.java | 21 +++++++-- .../RedisQueueMessageDrivenEndpointTests.java | 25 +++++------ 6 files changed, 91 insertions(+), 62 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 7f806518cf..ae09b004cc 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 0f0e00bcb0..4f57c43248 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; @@ -51,7 +52,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"; @@ -65,24 +67,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; /** @@ -92,7 +94,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()); @@ -109,12 +111,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}. @@ -146,11 +154,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(); errorHandler.setBeanFactory(getBeanFactory()); errorHandler.setDefaultErrorChannel(getErrorChannel()); @@ -168,8 +176,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()); @@ -178,7 +186,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 7ecded6029..cb935584a5 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 @@ -20,6 +20,7 @@ import java.util.Optional; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; @@ -40,7 +41,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; @@ -57,7 +57,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; @@ -65,19 +66,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; @@ -85,8 +88,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 @@ -94,7 +95,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()); @@ -107,8 +108,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; } /** @@ -116,8 +125,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) { @@ -144,12 +152,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; } @@ -178,7 +180,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && beanFactory != null) { MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(ChannelResolverUtils.getChannelResolver(beanFactory)); - 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 45cd9cbe56..206f56a21b 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 @@ -46,19 +46,19 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator(); - private static final RedisSerializer stringSerializer = new StringRedisSerializer(); + private static final RedisSerializer stringSerializer = new StringRedisSerializer(); private final RedisTemplate template = new RedisTemplate<>(); 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"); @@ -70,6 +70,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 5c622bb3f0..7f1f534752 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 f9af711815..883b72cc97 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 @@ -63,8 +63,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 @@ -74,8 +74,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * * @since 3.0 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) @DirtiesContext public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @@ -95,7 +94,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3014Default() { - String queueName = "si.test.redisQueueInboundChannelAdapterTests"; RedisTemplate redisTemplate = new RedisTemplate<>(); @@ -118,6 +116,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(); @@ -138,8 +137,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3014ExpectMessageTrue() { - - final String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; + String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; RedisTemplate redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(this.connectionFactory); @@ -188,7 +186,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Test @RedisAvailable public void testInt3017IntegrationInbound() { - String payload = new Date().toString(); RedisTemplate redisTemplate = new StringRedisTemplate(); @@ -222,9 +219,9 @@ 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 redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.setEnableDefaultSerializer(false); redisTemplate.setKeySerializer(new StringRedisSerializer()); @@ -257,9 +254,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(); assertThat(executorService.awaitTermination(20, TimeUnit.SECONDS)).isTrue(); @@ -309,7 +306,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()); @@ -331,7 +328,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @RedisAvailable @SuppressWarnings("unchecked") public void testInt3932ReadFromLeft() { - String queueName = "si.test.redisQueueInboundChannelAdapterTests3932"; RedisTemplate redisTemplate = new RedisTemplate<>(); @@ -354,6 +350,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);