INT-4379: JMS OG Shutdown reply container on stop

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

- shutdown the container when the gateway is stopped

Also, improve test suite

- at the end of the tests, hundreds of threads are running, some caused by the above
  but others because `TaskExecutor`s are not shut down
- reduce the number of iterations in the JMS pipeline tests to speed things up
- change more tests to extend `ActiveMQMultiContextTests`, to keep a single broker up

__cherry-pick to 4.3.x__

# Conflicts:
#	spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
#	spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java
#	spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java
#	spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java
#	spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java
#	spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -156,6 +156,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 
 	private volatile long idleReplyContainerTimeout;
 
+	private volatile boolean wasStopped;
+
 	private ScheduledFuture<?> idleTask;
 
 	/**
@@ -676,6 +678,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 				if (this.replyContainer != null) {
 					TaskScheduler taskScheduler = getTaskScheduler();
 					if (this.idleReplyContainerTimeout <= 0) {
+						if (this.wasStopped) {
+							this.replyContainer.initialize();
+							this.wasStopped = false;
+						}
 						this.replyContainer.start();
 					}
 					else {
@@ -695,7 +701,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 	public void stop() {
 		synchronized (this.lifeCycleMonitor) {
 			if (this.replyContainer != null) {
-				this.replyContainer.stop();
+				this.replyContainer.shutdown();
+				this.wasStopped = true;
 				this.deleteDestinationIfTemporary(this.replyContainer.getDestination());
 				if (this.reaper != null) {
 					this.reaper.cancel(false);

spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java

@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -69,13 +70,11 @@ import org.springframework.util.ObjectUtils;
  * @author Artem Bilan
  * @since 2.2.4
  */
-public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
+public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	public JmsOutboundGatewayTests() {
-		super("org.springframework.integration", "org.springframework.jms", "org.apache");
-	}
+
 
 	@Test
 	public void testContainerBeanNameWhenNoGatewayBeanName() {
@@ -100,16 +99,11 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		gateway.setUseReplyContainer(true);
 		ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties();
 		final List<Throwable> errors = new ArrayList<Throwable>();
+		ExecutorService exec = Executors.newFixedThreadPool(10);
 		ErrorHandlingTaskExecutor errorHandlingTaskExecutor =
-				new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() {
-
-					@Override
-					public void handleError(Throwable t) {
-						logger.info("Error:", t);
-						errors.add(t);
-						throw new RuntimeException(t);
-					}
-
+				new ErrorHandlingTaskExecutor(exec, t -> {
+					errors.add(t);
+					throw new RuntimeException(t);
 				});
 		replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor);
 		replyContainerProperties.setRecoveryInterval(100L);
@@ -173,6 +167,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		}
 		finally {
 			gateway.stop();
+			exec.shutdownNow();
 		}
 	}
 
@@ -192,13 +187,8 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		gateway.setReceiveTimeout(60000);
 		gateway.afterPropertiesSet();
 		gateway.start();
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-
-			@Override
-			public void run() {
-				gateway.handleMessage(new GenericMessage<String>("foo"));
-			}
-		});
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
 		CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
 				new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
 		JmsTemplate template = new JmsTemplate(connectionFactory2);
@@ -224,6 +214,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		gateway.stop();
 		connectionFactory1.destroy();
 		connectionFactory2.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -243,13 +234,8 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		gateway.setCorrelationKey("JMSCorrelationID");
 		gateway.afterPropertiesSet();
 		gateway.start();
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-
-			@Override
-			public void run() {
-				gateway.handleMessage(new GenericMessage<String>("foo"));
-			}
-		});
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
 		CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
 				new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
 		JmsTemplate template = new JmsTemplate(connectionFactory2);
@@ -276,6 +262,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
 		gateway.stop();
 		connectionFactory1.destroy();
 		connectionFactory2.destroy();
+		exec.shutdownNow();
 	}
 
 }

spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -36,11 +36,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * //INT-2275
  *
  * @author Artem Bilan
+ * @author Gary Russell
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration
 @DirtiesContext
-public class JmsOutboundInsideChainTests {
+public class JmsOutboundInsideChainTests extends ActiveMQMultiContextTests {
 
 	@Autowired
 	private MessageChannel outboundChainChannel;

spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -22,6 +22,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -37,12 +38,13 @@ import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.junit.Ignore;
 import org.junit.Test;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.context.IntegrationContextUtils;
-import org.springframework.messaging.support.GenericMessage;
 import org.springframework.jms.connection.CachingConnectionFactory;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.core.MessageCreator;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
@@ -57,14 +59,15 @@ public class OutboundGatewayConnectionTests {
 
 	private Destination replyQueue1 = new ActiveMQQueue("reply1");
 
-	@Test @Ignore // need a more reliable stop/start for AMQ
+	@Test
+	@Ignore // need a more reliable stop/start for AMQ
 	public void testContainerWithDestBrokenConnection() throws Exception {
 		BeanFactory beanFactory = mock(BeanFactory.class);
 		when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		BrokerService broker = new BrokerService();
@@ -82,15 +85,14 @@ public class OutboundGatewayConnectionTests {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -116,15 +118,13 @@ public class OutboundGatewayConnectionTests {
 
 		final CountDownLatch latch3 = new CountDownLatch(1);
 		final CountDownLatch latch4 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			public void run() {
-				latch3.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch4.countDown();
-				}
+		exec.execute(() -> {
+			latch3.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
+			}
+			finally {
+				latch4.countDown();
 			}
 		});
 		assertTrue(latch3.await(10, TimeUnit.SECONDS));
@@ -147,6 +147,7 @@ public class OutboundGatewayConnectionTests {
 		broker.stop();
 
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 }

spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java

@@ -23,6 +23,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -40,7 +41,6 @@ import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.context.IntegrationContextUtils;
-import org.springframework.integration.test.support.LogAdjustingTestSupport;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.jms.JmsException;
 import org.springframework.jms.connection.CachingConnectionFactory;
@@ -56,7 +56,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  * @since 2.2
  *
  */
-public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
+public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests {
 
 	private static Destination requestQueue1 = new ActiveMQQueue("request1");
 
@@ -85,7 +85,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		ConnectionFactory connectionFactory = getConnectionFactory();
@@ -99,16 +99,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -130,6 +128,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdown();
 	}
 
 	@Test
@@ -139,7 +138,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -151,16 +150,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -183,6 +180,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -192,7 +190,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -205,16 +203,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -236,6 +232,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -245,7 +242,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -257,16 +254,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -289,6 +284,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -298,7 +294,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -311,16 +307,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -342,6 +336,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -352,7 +347,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -363,16 +358,14 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		final AtomicReference<Object> reply = new AtomicReference<Object>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
 		final CountDownLatch latch2 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				latch1.countDown();
-				try {
-					reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
-				}
-				finally {
-					latch2.countDown();
-				}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			latch1.countDown();
+			try {
+				reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo")));
+			}
+			finally {
+				latch2.countDown();
 			}
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -395,6 +388,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 
 		gateway.stop();
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -404,7 +398,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
 		scheduler.initialize();
 		when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
-			.thenReturn(scheduler);
+				.thenReturn(scheduler);
 		final JmsOutboundGateway gateway = new JmsOutboundGateway();
 		gateway.setBeanFactory(beanFactory);
 		gateway.setConnectionFactory(getConnectionFactory());
@@ -417,33 +411,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		gateway.setReceiveTimeout(20000);
 		gateway.afterPropertiesSet();
 		gateway.start();
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-			@Override
-			public void run() {
-				JmsTemplate template = new JmsTemplate();
-				template.setConnectionFactory(getConnectionFactory());
-				template.setReceiveTimeout(20000);
-				receiveAndSend(template);
-				receiveAndSend(template);
-			}
-
-			private void receiveAndSend(JmsTemplate template) {
-				javax.jms.Message request = template.receive(requestQueue7);
-				final javax.jms.Message jmsReply = request;
-				try {
-					template.send(request.getJMSReplyTo(), new MessageCreator() {
-
-						@Override
-						public Message createMessage(Session session) throws JMSException {
-							return jmsReply;
-						}
-					});
-				}
-				catch (JmsException e) {
-				}
-				catch (JMSException e) {
-				}
-			}
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			JmsTemplate template = new JmsTemplate();
+			template.setConnectionFactory(getConnectionFactory());
+			template.setReceiveTimeout(20000);
+			receiveAndSend(template);
+			receiveAndSend(template);
 		});
 
 		assertNotNull(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
@@ -460,6 +434,17 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
 		gateway.stop();
 		assertFalse(container.isRunning());
 		scheduler.destroy();
+		exec.shutdownNow();
+	}
+
+	private void receiveAndSend(JmsTemplate template) {
+		javax.jms.Message request = template.receive(requestQueue7);
+		final javax.jms.Message jmsReply = request;
+		try {
+			template.send(request.getJMSReplyTo(), session -> jmsReply);
+		}
+		catch (JmsException | JMSException e) {
+		}
 	}
 
 	private ConnectionFactory getConnectionFactory() {

spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -28,9 +28,11 @@ import static org.mockito.Mockito.verify;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Session;
@@ -61,7 +63,7 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Gunnar Hillert
  * @author Artem Bilan
  */
-public class PollableJmsChannelTests {
+public class PollableJmsChannelTests extends ActiveMQMultiContextTests {
 
 	private ActiveMQConnectionFactory connectionFactory;
 
@@ -190,14 +192,10 @@ public class PollableJmsChannelTests {
 		assertTrue(sent1);
 		final AtomicReference<javax.jms.Message> message = new AtomicReference<javax.jms.Message>();
 		final CountDownLatch latch1 = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-
-			@Override
-			public void run() {
-				message.set(receiver.receive(queue));
-				latch1.countDown();
-			}
-
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
+			message.set(receiver.receive(queue));
+			latch1.countDown();
 		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
 		assertNotNull(message.get());
@@ -208,20 +206,16 @@ public class PollableJmsChannelTests {
 		final CountDownLatch latch2 = new CountDownLatch(1);
 		boolean sent2 = channel.send(MessageBuilder.withPayload("foo").setPriority(6).build());
 		assertTrue(sent2);
-		Executors.newSingleThreadExecutor().execute(new Runnable() {
-
-			@Override
-			public void run() {
-				message.set(receiver.receive(queue));
-				latch2.countDown();
-			}
-
+		exec.execute(() -> {
+			message.set(receiver.receive(queue));
+			latch2.countDown();
 		});
 		assertTrue(latch2.await(10, TimeUnit.SECONDS));
 		assertNotNull(message.get());
 		assertEquals(6, message.get().getJMSPriority());
 		assertTrue(message.get().getJMSExpiration() <= System.currentTimeMillis() + ttl);
 		assertTrue(message.get().toString().contains("persistent = false"));
+		exec.shutdownNow();
 	}
 
 	@Test

spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -70,7 +70,7 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Artem Bilan
  * @since 2.0
  */
-public class SubscribableJmsChannelTests {
+public class SubscribableJmsChannelTests extends ActiveMQMultiContextTests {
 
 	private static final int TIMEOUT = 30000;
 

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -20,7 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -28,6 +28,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
 import org.apache.log4j.LogManager;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -39,6 +40,7 @@ import org.springframework.integration.jms.ActiveMQMultiContextTests;
 import org.springframework.integration.jms.config.ActiveMqTestUtils;
 import org.springframework.integration.test.support.LongRunningIntegrationTest;
 import org.springframework.messaging.support.GenericMessage;
+
 /**
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -46,7 +48,7 @@ import org.springframework.messaging.support.GenericMessage;
  */
 public class PipelineJmsTests extends ActiveMQMultiContextTests {
 
-	private final Executor executor = Executors.newFixedThreadPool(30);
+	private final ExecutorService executor = Executors.newFixedThreadPool(30);
 
 	private static final Log logger = LogFactory.getLog(PipelineJmsTests.class);
 
@@ -58,7 +60,12 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
 		LogManager.getLogger(getClass()).setLevel(Level.INFO);
 	}
 
-	int requests = 50;
+	@After
+	public void tearDown() {
+		this.executor.shutdownNow();
+	}
+
+	int requests = 5;
 
 	/**
 	 * jms:out -> jms:in -> randomTimeoutProcess ->
@@ -162,6 +169,7 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
 			for (int i = 0; i < requests; i++) {
 				final int y = i;
 				executor.execute(new Runnable() {
+
 					@Override
 					public void run() {
 						try {
@@ -189,7 +197,7 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
 			logger.info("Failure: " + failureCounter.get());
 			// technically all we care that its > 0,
 			// but reality of this test it has to be something more then 0
-			assertTrue(successCounter.get() > 10);
+			assertTrue(successCounter.get() > 1);
 			assertEquals(0, failureCounter.get());
 			assertEquals(requests, successCounter.get() + timeoutCounter.get());
 			context.close();

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -31,6 +31,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
 import org.apache.log4j.LogManager;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -43,6 +44,7 @@ import org.springframework.integration.jms.config.ActiveMqTestUtils;
 import org.springframework.integration.test.support.LongRunningIntegrationTest;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
+
 /**
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -51,7 +53,7 @@ import org.springframework.messaging.support.GenericMessage;
  */
 public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests {
 
-	private final Executor executor = Executors.newFixedThreadPool(30);
+	private final ExecutorService executor = Executors.newFixedThreadPool(30);
 
 	private static final Log logger = LogFactory.getLog(PipelineJmsTests.class);
 
@@ -63,7 +65,12 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 		LogManager.getLogger(getClass()).setLevel(Level.INFO);
 	}
 
-	int requests = 50;
+	@After
+	public void tearDown() {
+		this.executor.shutdownNow();
+	}
+
+	int requests = 5;
 
 	/**
 	 * jms:out(reply-destination-name="pipeline01-01") -> jms:in -> randomTimeoutProcess ->
@@ -175,6 +182,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 			for (int i = 1000000; i < 1000000 + requests * 100000; i += 100000) {
 				final int y = i;
 				executor.execute(new Runnable() {
+
 					@Override
 					public void run() {
 						try {
@@ -197,7 +205,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 			assertTrue(latch.await(120, TimeUnit.SECONDS));
 			// technically all we care that its > 0,
 			// but reality of this test it has to be something more then 0
-			assertTrue(successCounter.get() > 10);
+			assertTrue(successCounter.get() > 1);
 			assertEquals(0, failureCounter.get());
 			assertEquals(requests, successCounter.get() + timeoutCounter.get());
 			return timeoutCounter.get();

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -19,11 +19,9 @@ package org.springframework.integration.jms.request_reply;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -56,6 +54,7 @@ import org.springframework.jms.listener.SessionAwareMessageListener;
 import org.springframework.jms.support.converter.SimpleMessageConverter;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.support.GenericMessage;
+
 /**
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -73,7 +72,8 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 	public void messageCorrelationBasedOnRequestMessageId() throws Exception {
 		ActiveMqTestUtils.prepare();
 
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass());
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass());
 		RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
 		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
 		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
@@ -187,7 +187,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		BrokerService broker = new BrokerService();
 		broker.setPersistent(false);
 		broker.setUseJmx(false);
-		broker.setTransportConnectorURIs(new String[]{"tcp://localhost:61623"});
+		broker.setTransportConnectorURIs(new String[] { "tcp://localhost:61623" });
 		broker.setDeleteAllMessagesOnStartup(true);
 		broker.start();
 
@@ -223,7 +223,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		ClassPathXmlApplicationContext context =
 				new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass());
 		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
-		Executor executor = Executors.newFixedThreadPool(10);
+		ExecutorService executor = Executors.newFixedThreadPool(10);
 		final int testNumbers = 100;
 		final CountDownLatch latch = new CountDownLatch(testNumbers);
 		final AtomicInteger failures = new AtomicInteger();
@@ -232,6 +232,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		for (int i = 0; i < testNumbers; i++) {
 			final int y = i;
 			executor.execute(new Runnable() {
+
 				@Override
 				public void run() {
 					try {
@@ -266,6 +267,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		assertEquals(0, failures.get());
 		assertEquals(0, timeouts.get());
 		context.close();
+		executor.shutdownNow();
 	}
 
 	private void print(AtomicInteger failures, AtomicInteger timeouts, AtomicInteger missmatches, long echangesProcessed) {
@@ -273,20 +275,10 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		logger.info(echangesProcessed + " exchanges processed");
 		logger.info("Failures: " + failures.get());
 		logger.info("Timeouts: " + timeouts.get());
-		logger.info("Missmatches: " + missmatches.get());
+		logger.info("Mismatches: " + missmatches.get());
 		logger.info("============================");
 	}
 
-	public static class MyRandomlySlowService {
-		Random random = new Random();
-		List<Integer> list = new ArrayList<Integer>();
-		public String secho(String value) throws Exception {
-			int i = random.nextInt(2000);
-			Thread.sleep(i);
-			return value;
-		}
-	}
-
 	private Object extractPayload(Message jmsMessage) {
 		try {
 			return converter.fromMessage(jmsMessage);
@@ -297,4 +289,17 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
 		}
 		return null;
 	}
+
+	public static class MyRandomlySlowService {
+
+		Random random = new Random();
+
+		public String echo(String value) throws Exception {
+			int i = random.nextInt(2000);
+			Thread.sleep(i);
+			return value;
+		}
+
+	}
+
 }