From 4cf0e2e673a77828851b0d1834d9a692b5d84e4b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 5 Nov 2013 12:21:47 +0200 Subject: [PATCH] INT-3196: RedisQueueMDE: Add RecoveryInterval Catch `RedisSystemException` in the `ListenerTask` and sleep current Thread with `recoveryInterval` if the Endpoint is `active` before the next `restart()` JIRA: https://jira.springsource.org/browse/INT-3196 RedisConnFailure & RedisSys Exceptions recovery Introduce `RedisIntegrationEvent`s * Catch all `Exception`s on `this.boundListOperations.rightPop` * Log them and send within `RedisExceptionEvent` * Mark some fields with `volatile` --- ...RedisQueueInboundChannelAdapterParser.java | 35 ++-------- .../redis/event/RedisExceptionEvent.java | 30 ++++++++ .../redis/event/RedisIntegrationEvent.java | 37 ++++++++++ .../integration/redis/event/package-info.java | 4 ++ .../RedisQueueMessageDrivenEndpoint.java | 62 +++++++++++++--- .../RedisQueueOutboundChannelAdapter.java | 2 +- .../RedisQueueMessageDrivenEndpointTests.java | 70 ++++++++++++++++++- 7 files changed, 197 insertions(+), 43 deletions(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java index ab0cf50cc3..18996384da 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java @@ -18,11 +18,10 @@ package org.springframework.integration.redis.config; import org.w3c.dom.Element; -import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint; import org.springframework.util.StringUtils; @@ -33,29 +32,11 @@ import org.springframework.util.StringUtils; * @author Artem Bilan * @since 3.0 */ -public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser { +public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapterParser { @Override - protected Class getBeanClass(Element element) { - return RedisQueueMessageDrivenEndpoint.class; - } - - @Override - protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) - throws BeanDefinitionStoreException { - String id = element.getAttribute("id"); - if (!element.hasAttribute("channel")) { - // the created channel will get the 'id', so the adapter's bean name includes a suffix - id = id + ".adapter"; - } - else if (!StringUtils.hasText(id)) { - id = parserContext.getReaderContext().generateBeanName(definition); - } - return id; - } - - @Override - protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueMessageDrivenEndpoint.class); builder.addConstructorArgValue(element.getAttribute("queue")); String connectionFactory = element.getAttribute("connection-factory"); @@ -69,15 +50,9 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDef IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); - - String channelName = element.getAttribute("channel"); - if (!StringUtils.hasText(channelName)) { - channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext); - } builder.addPropertyReference("outputChannel", channelName); + return builder.getBeanDefinition(); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java new file mode 100644 index 0000000000..62115bfa6a --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java @@ -0,0 +1,30 @@ +/* + * Copyright 2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.event; + +/** + * @author Artem Bilan + * @since 3.0 + */ +@SuppressWarnings("serial") +public class RedisExceptionEvent extends RedisIntegrationEvent { + + public RedisExceptionEvent(Object source, Throwable cause) { + super(source, cause); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java new file mode 100644 index 0000000000..e2a91068d1 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java @@ -0,0 +1,37 @@ +/* + * Copyright 2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.event; + +import org.springframework.integration.event.IntegrationEvent; + +/** + * @author Artem Bilan + * @since 3.0 + * + */ +@SuppressWarnings("serial") +public abstract class RedisIntegrationEvent extends IntegrationEvent { + + public RedisIntegrationEvent(Object source) { + super(source); + } + + public RedisIntegrationEvent(Object source, Throwable cause) { + super(source, cause); + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java new file mode 100644 index 0000000000..3c4d6dec86 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java @@ -0,0 +1,4 @@ +/** + * Events generated by the redis module + */ +package org.springframework.integration.redis.event; 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 1c33335f27..36d7f85a2c 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 @@ -18,8 +18,9 @@ package org.springframework.integration.redis.inbound; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundListOperations; import org.springframework.data.redis.core.RedisTemplate; @@ -31,6 +32,7 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.redis.event.RedisExceptionEvent; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.integration.util.ErrorHandlingTaskExecutor; @@ -46,13 +48,17 @@ import org.springframework.util.Assert; * @since 3.0 */ @ManagedResource -public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { +public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware { public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + public static final long DEFAULT_RECOVERY_INTERVAL = 5000; + private final BoundListOperations boundListOperations; - private MessageChannel errorChannel; + private volatile ApplicationEventPublisher applicationEventPublisher; + + private volatile MessageChannel errorChannel; private volatile Executor taskExecutor; @@ -62,6 +68,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; + private volatile boolean active; private volatile boolean listening; @@ -81,6 +89,11 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { this.boundListOperations = template.boundListOps(queueName); } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + public void setSerializer(RedisSerializer serializer) { this.serializer = serializer; } @@ -129,6 +142,10 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { this.errorChannel = errorChannel; } + public void setRecoveryInterval(long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + @Override protected void onInit() { super.onInit(); @@ -160,13 +177,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { try { value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); } - catch (RedisSystemException e) { - if (this.active) { - throw e; - } - else { - logger.error(e); - } + catch (Exception e) { + logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e); + this.listening = false; + this.sleepBeforeRecoveryAttempt(); + this.publishException(e); + return; } if (value != null) { @@ -200,6 +216,32 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { } } + /** + * Sleep according to the specified recovery interval. + * Called between recovery attempts. + */ + private void sleepBeforeRecoveryAttempt() { + if (this.recoveryInterval > 0) { + try { + Thread.sleep(this.recoveryInterval); + } + catch (InterruptedException e) { + logger.debug("Thread interrupted while sleeping the recovery interval"); + } + } + } + + private void publishException(Exception e) { + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e)); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("No application event publisher for exception: " + e.getMessage()); + } + } + } + private void restart() { this.taskExecutor.execute(new ListenerTask()); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java index 2ff459ec78..b25e058888 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java @@ -43,7 +43,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp private final Expression queueNameExpression; - private EvaluationContext evaluationContext; + private volatile EvaluationContext evaluationContext; private volatile boolean extractPayload = 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 5a1c33c8a5..ec28d11af0 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 @@ -18,9 +18,13 @@ package org.springframework.integration.redis.inbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Date; +import java.util.List; import java.util.UUID; import org.hamcrest.Matchers; @@ -29,7 +33,13 @@ import org.junit.runner.RunWith; import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; @@ -40,7 +50,9 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.event.IntegrationEvent; import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.redis.event.RedisExceptionEvent; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.MessageBuilder; @@ -68,7 +80,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Autowired private PollableChannel symmetricalOutputChannel; - @Test @RedisAvailable @SuppressWarnings("unchecked") @@ -158,7 +169,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertThat(((Exception) receive.getPayload()).getCause().getMessage(), Matchers.containsString("java.lang.String cannot be cast to org.springframework.integration.Message")); - endpoint.stop(); } @@ -194,4 +204,60 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload, receive.getPayload()); } + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3196Recovery() throws Exception { + String queueName = "test.si.Int3196Recovery"; + QueueChannel channel = new QueueChannel(); + + final List exceptionEvents = new ArrayList(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); + endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); + endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + exceptionEvents.add(event); + } + }); + endpoint.setOutputChannel(channel); + endpoint.setReceiveTimeout(100); + endpoint.setRecoveryInterval(200); + endpoint.afterPropertiesSet(); + endpoint.start(); + + ((DisposableBean) this.connectionFactory).destroy(); + + Thread.sleep(300); + + assertThat(exceptionEvents.size(), Matchers.greaterThan(0)); + for (ApplicationEvent exceptionEvent : exceptionEvents) { + assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class)); + assertSame(endpoint, exceptionEvent.getSource()); + assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(), + Matchers.isIn(Arrays.> asList(RedisSystemException.class, RedisConnectionFailureException.class))); + } + + ((InitializingBean) this.connectionFactory).afterPropertiesSet(); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest()); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + String payload = "testing"; + + redisTemplate.boundListOps(queueName).leftPush(payload); + + Message receive = channel.receive(1000); + assertNotNull(receive); + assertEquals(payload, receive.getPayload()); + + endpoint.stop(); + } + }