INT-3486: Poller: Prevent Interrupt StackTrace

JIRA: https://jira.spring.io/browse/INT-3486

INT-3486a: PR comments

spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java

@@ -28,8 +28,7 @@ public interface MessageSource<T> {
 	/**
 	 * Retrieve the next available message from this source.
 	 * Returns <code>null</code> if no message is available.
-	 *
-	 * @return The messasge or null.
+	 * @return The message or null.
 	 */
 	Message<T> receive();
 

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()){

spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java

@@ -160,11 +160,11 @@ public class SimplePool<T> implements Pool<T> {
 				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();
 		}

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<Object> ms = new MessageSource<Object>() {
+			@Override
+			public Message<Object> 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<String> {
 
 		public Message<String> receive() {