From cda4a990230bb582f2717e4f0c75b4177a128fe8 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 19 Aug 2014 00:28:12 +0300 Subject: [PATCH] INT-3486: Poller: Prevent Interrupt StackTrace JIRA: https://jira.spring.io/browse/INT-3486 INT-3486a: PR comments --- .../integration/core/MessageSource.java | 3 +- .../endpoint/AbstractPollingEndpoint.java | 16 +++- .../integration/util/SimplePool.java | 4 +- ...PollingChannelAdapterFactoryBeanTests.java | 84 +++++++++++++++++-- 4 files changed, 96 insertions(+), 11 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java index 071e286a0d..2a7f9adb6b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java @@ -28,8 +28,7 @@ public interface MessageSource { /** * Retrieve the next available message from this source. * Returns null if no message is available. - * - * @return The messasge or null. + * @return The message or null. */ Message receive(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 58aad68402..c5b8edd7f5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -187,7 +187,21 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private boolean doPoll() { IntegrationResourceHolder holder = this.bindResourceHolderIfNecessary( this.getResourceKey(), this.getResourceToBind()); - Message message = this.receiveMessage(); + Message message = null; + try { + message = this.receiveMessage(); + } + catch (Exception e) { + if (Thread.interrupted()) { + if (logger.isDebugEnabled()) { + logger.debug("Poll interrupted - during stop()? : " + e.getMessage()); + } + return false; + } + else { + throw (RuntimeException) e; + } + } boolean result; if (message == null) { if (this.logger.isDebugEnabled()){ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index de4d565ae7..ca0c58deb5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -160,11 +160,11 @@ public class SimplePool implements Pool { permitted = this.permits.tryAcquire(this.waitTimeout, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { - logger.error("Interrupted awaiting a pooled resource."); Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted awaiting a pooled resource", e); } if (!permitted) { - throw new IllegalStateException("Timed out while waiting to aquire a pool entry."); + throw new IllegalStateException("Timed out while waiting to acquire a pool entry."); } return doGetItem(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java index 4c00eabb8c..7ad0b12f5e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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. @@ -18,33 +18,49 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.contains; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import org.springframework.messaging.Message; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageSource; -import org.springframework.messaging.support.GenericMessage; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.test.util.TestUtils.TestApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.ClassUtils; /** * @author Mark Fisher + * @author Artem Bilan */ public class SourcePollingChannelAdapterFactoryBeanTests { @@ -102,7 +118,7 @@ public class SourcePollingChannelAdapterFactoryBeanTests { pollerMetadata.setMaxMessagesPerPoll(1); final AtomicInteger count = new AtomicInteger(); final MethodInterceptor txAdvice = mock(MethodInterceptor.class); - adviceChain.add(new MethodInterceptor() { + adviceChain.add(new MethodInterceptor() { public Object invoke(MethodInvocation invocation) throws Throwable { count.incrementAndGet(); return invocation.proceed(); @@ -111,10 +127,10 @@ public class SourcePollingChannelAdapterFactoryBeanTests { when(txAdvice.invoke(Mockito.any(MethodInvocation.class))).thenAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { count.incrementAndGet(); - return ((MethodInvocation)invocation.getArguments()[0]).proceed(); + return ((MethodInvocation) invocation.getArguments()[0]).proceed(); } }); - + pollerMetadata.setAdviceChain(adviceChain); factoryBean.setPollerMetadata(pollerMetadata); factoryBean.setAutoStartup(true); @@ -127,6 +143,62 @@ public class SourcePollingChannelAdapterFactoryBeanTests { assertTrue("adviceChain was not applied", adviceApplied.get()); } + @Test + public void testInterrupted() throws Exception { + final CountDownLatch startLatch = new CountDownLatch(1); + + MessageSource ms = new MessageSource() { + @Override + public Message receive() { + startLatch.countDown(); + try { + Thread.sleep(10000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted awaiting stopLatch", e); + } + return null; + } + }; + + SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter(); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.setWaitForTasksToCompleteOnShutdown(true); + taskScheduler.setAwaitTerminationSeconds(1); + taskScheduler.afterPropertiesSet(); + pollingChannelAdapter.setTaskScheduler(taskScheduler); + + MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(); + Log errorHandlerLogger = TestUtils.getPropertyValue(errorHandler, "logger", Log.class); + errorHandlerLogger = spy(errorHandlerLogger); + DirectFieldAccessor dfa = new DirectFieldAccessor(errorHandler); + dfa.setPropertyValue("logger", errorHandlerLogger); + pollingChannelAdapter.setErrorHandler(errorHandler); + + pollingChannelAdapter.setSource(ms); + pollingChannelAdapter.setOutputChannel(new NullChannel()); + pollingChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + pollingChannelAdapter.afterPropertiesSet(); + + Log adapterLogger = TestUtils.getPropertyValue(pollingChannelAdapter, "logger", Log.class); + adapterLogger = spy(adapterLogger); + when(adapterLogger.isDebugEnabled()).thenReturn(true); + + dfa = new DirectFieldAccessor(pollingChannelAdapter); + dfa.setPropertyValue("logger", adapterLogger); + + pollingChannelAdapter.start(); + + assertTrue(startLatch.await(10, TimeUnit.SECONDS)); + pollingChannelAdapter.stop(); + + taskScheduler.shutdown(); + + verifyZeroInteractions(errorHandlerLogger); + verify(adapterLogger).debug(contains("Poll interrupted - during stop()?")); + } + private static class TestSource implements MessageSource { public Message receive() {