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

Author: garyrussell

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 void start() {
 				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 void start() {
 	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 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 @@
  * @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 void testReplyContainerRecovery() throws Exception {
 		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 Message answer(InvocationOnMock invocation) throws Throwable {
 		}
 		finally {
 			gateway.stop();
+			exec.shutdownNow();
 		}
 	}
 
@@ -192,13 +187,8 @@ public void testConnectionBreakOnReplyMessageIdCorrelation() throws Exception {
 		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 Message createMessage(Session session) throws JMSException {
 		gateway.stop();
 		connectionFactory1.destroy();
 		connectionFactory2.destroy();
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -243,13 +234,8 @@ public void testConnectionBreakOnReplyCustomCorrelation() throws Exception {
 		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 Message createMessage(Session session) throws JMSException {
 		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 @@
  * //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.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.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 void testContainerWithDestBrokenConnection() throws Exception {
 		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 Message createMessage(Session session) throws JMSException {
 
 		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 Message createMessage(Session session) throws JMSException {
 		broker.stop();
 
 		scheduler.destroy();
+		exec.shutdownNow();
 	}
 
 }