Shutdown test executors in -core

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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.
@@ -77,7 +77,8 @@ protected void afterRelease(MessageGroup group, Collection<Message<?>> completed
 		/*
 		 * Runs "reap" when group 'bar' is in completion
 		 */
-		Executors.newSingleThreadExecutor().execute(() -> {
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
 			try {
 				waitReapStartLatch.await(10, TimeUnit.SECONDS);
 			}
@@ -145,6 +146,7 @@ protected void afterRelease(MessageGroup group, Collection<Message<?>> completed
 		assertEquals(1, ((MessageGroup) outputMessages.get(1).getPayload()).size()); // 'qux'
 
 		assertNull(discards.receive(0));
+		exec.shutdownNow();
 	}
 
 	@Test // INT-2833

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -272,7 +272,7 @@ public void testShouldNotSendPartialResultOnTimeoutByDefault() throws Interrupte
 		Message<?> message = createMessage(3, "ABC", 2, 1, replyChannel, null);
 		this.aggregator.handleMessage(message);
 		this.store.expireMessageGroups(-10000);
-		Message<?> reply = replyChannel.receive(1000);
+		Message<?> reply = replyChannel.receive(0);
 		assertNull("No message should have been sent normally", reply);
 		Message<?> discardedMessage = discardChannel.receive(1000);
 		assertNotNull("A message should have been discarded", discardedMessage);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -132,6 +132,7 @@ public void testRequestBeforeReply() throws Exception {
 		assertEquals("bar", result.get(1));
 		assertEquals(0, suspensions.size());
 		assertEquals(0, inProcess.size());
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -141,8 +142,8 @@ public void testReplyBeforeRequest() throws Exception {
 		handler.setOutputChannel(outputChannel);
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
-		Executors.newSingleThreadExecutor()
-				.execute(() -> handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()));
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build()));
 		Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
 		int n = 0;
 		while (n++ < 100 && suspensions.size() == 0) {
@@ -156,6 +157,7 @@ public void testReplyBeforeRequest() throws Exception {
 		assertEquals("foo", result.get(0));
 		assertEquals("bar", result.get(1));
 		assertEquals(0, suspensions.size());
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -169,7 +171,8 @@ public void testLateReply() throws Exception {
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
 		final CountDownLatch latch = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(() -> {
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
 			handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
 			latch.countDown();
 		});
@@ -190,6 +193,7 @@ public void testLateReply() throws Exception {
 		assertSame(discard, triggerMessage);
 		handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
 		assertEquals(0, suspensions.size());
+		exec.shutdownNow();
 	}
 
 	@Test
@@ -218,7 +222,8 @@ public void testExceptionReply() throws Exception {
 		handler.afterPropertiesSet();
 		final AtomicReference<Exception> exception = new AtomicReference<Exception>();
 		final CountDownLatch latch = new CountDownLatch(1);
-		Executors.newSingleThreadExecutor().execute(() -> {
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.execute(() -> {
 			try {
 				handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
 			}
@@ -238,6 +243,7 @@ public void testExceptionReply() throws Exception {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertSame(exc, exception.get().getCause());
 		assertEquals(0, suspensions.size());
+		exec.shutdownNow();
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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.
@@ -27,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 
@@ -89,8 +90,9 @@ public void shouldNotDropMessageOrBlockSendingThread() {
 		barrier.setReleaseStrategy(trackingReleaseStrategy);
 		final CountDownLatch start = new CountDownLatch(1);
 		final CountDownLatch sent = new CountDownLatch(200);
+		ExecutorService exec = Executors.newSingleThreadExecutor();
 		for (int i = 0; i < 200; i++) {
-			sendAsynchronously(barrier, testMessage(), start, sent);
+			sendAsynchronously(barrier, testMessage(), start, sent, exec);
 		}
 		start.countDown();
 
@@ -106,10 +108,12 @@ public void shouldNotDropMessageOrBlockSendingThread() {
 			trackingReleaseStrategy.release("foo");
 			assertThat((barrier.receive()), is(notNullValue()));
 		}
+		exec.shutdownNow();
 	}
 
-	private void sendAsynchronously(final MessageHandler handler, final Message<Object> message, final CountDownLatch start, final CountDownLatch sent) {
-		Executors.newSingleThreadExecutor().execute(() -> {
+	private void sendAsynchronously(final MessageHandler handler, final Message<Object> message,
+			final CountDownLatch start, final CountDownLatch sent, ExecutorService exec) {
+		exec.execute(() -> {
 			try {
 				start.await();
 			}
@@ -119,7 +123,6 @@ private void sendAsynchronously(final MessageHandler handler, final Message<Obje
 			handler.handleMessage(message);
 			sent.countDown();
 		});
-
 	}
 
 	private Message<Object> testMessage() {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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,6 +28,7 @@
 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;
 
@@ -146,7 +147,8 @@ public void shouldNotPruneWhileCompleting() throws Exception {
 		handler.handleMessage(message1);
 		bothMessagesHandled.countDown();
 		storedMessages.add(message1);
-		Executors.newSingleThreadExecutor().submit(() -> {
+		ExecutorService exec = Executors.newSingleThreadExecutor();
+		exec.submit(() -> {
 			handler.handleMessage(message2);
 			storedMessages.add(message2);
 			bothMessagesHandled.countDown();
@@ -155,6 +157,7 @@ public void shouldNotPruneWhileCompleting() throws Exception {
 		assertTrue(bothMessagesHandled.await(10, TimeUnit.SECONDS));
 
 		assertEquals(0, store.expireMessageGroups(10000));
+		exec.shutdownNow();
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java

@@ -92,7 +92,7 @@ public void validateSequenceSizeHasNoAffectCustomCorrelator() throws Exception {
 			});
 		}
 
-		assertTrue("Sends failed to complete: " + latch.getCount() + " remain", latch.await(60, TimeUnit.SECONDS));
+		assertTrue("Sends failed to complete: " + latch.getCount() + " remain", latch.await(120, TimeUnit.SECONDS));
 
 		Message<?> message = resultChannel.receive(1000);
 		int counter = 0;

spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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,8 +20,7 @@
 import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -36,6 +35,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
@@ -128,8 +128,8 @@ public ProxyFactoryBean publishSubscribeChannel() {
 		}
 
 		@Bean
-		public ExecutorService executor() {
-			return Executors.newCachedThreadPool();
+		public Executor executor() {
+			return new ThreadPoolTaskExecutor();
 		}
 
 		private ProxyFactoryBean createProxyFactory(MessageChannel channel) {

spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.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.
@@ -32,6 +32,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -76,8 +77,9 @@ public void verifyDifferentThread() throws Exception {
 	@Test
 	public void roundRobinLoadBalancing() throws Exception {
 		int numberOfMessages = 11;
-		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
-				Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
+		ScheduledExecutorService exec = Executors
+				.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
+		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
 		ExecutorChannel channel = new ExecutorChannel(
 				taskExecutor, new RoundRobinLoadBalancingStrategy());
 		CountDownLatch latch = new CountDownLatch(numberOfMessages);
@@ -104,13 +106,15 @@ public void roundRobinLoadBalancing() throws Exception {
 		assertEquals(4, handler1.count.get());
 		assertEquals(4, handler2.count.get());
 		assertEquals(3, handler3.count.get());
+		exec.shutdownNow();
 	}
 
 	@Test
 	public void verifyFailoverWithLoadBalancing() throws Exception {
 		int numberOfMessages = 11;
-		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
-				Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
+		ScheduledExecutorService exec = Executors
+				.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
+		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
 		ExecutorChannel channel = new ExecutorChannel(
 				taskExecutor, new RoundRobinLoadBalancingStrategy());
 		CountDownLatch latch = new CountDownLatch(numberOfMessages);
@@ -138,13 +142,15 @@ public void verifyFailoverWithLoadBalancing() throws Exception {
 		assertEquals(0, handler2.count.get());
 		assertEquals(4, handler1.count.get());
 		assertEquals(7, handler3.count.get());
+		exec.shutdownNow();
 	}
 
 	@Test
 	public void verifyFailoverWithoutLoadBalancing() throws Exception {
 		int numberOfMessages = 11;
-		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(
-				Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-")));
+		ScheduledExecutorService exec = Executors
+				.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("test-"));
+		ConcurrentTaskExecutor taskExecutor = new ConcurrentTaskExecutor(exec);
 		ExecutorChannel channel = new ExecutorChannel(taskExecutor, null);
 		CountDownLatch latch = new CountDownLatch(numberOfMessages);
 		TestHandler handler1 = new TestHandler(latch);
@@ -169,6 +175,7 @@ public void verifyFailoverWithoutLoadBalancing() throws Exception {
 		assertEquals(0, handler1.count.get());
 		assertEquals(0, handler3.count.get());
 		assertEquals(numberOfMessages, handler2.count.get());
+		exec.shutdownNow();
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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.
@@ -32,6 +32,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -40,7 +41,7 @@
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import org.springframework.context.ApplicationContext;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -69,7 +70,7 @@ public class MixedDispatcherConfigurationScenarioTests {
 	@Mock
 	private List<Exception> exceptionRegistry;
 
-	private ApplicationContext ac;
+	private ConfigurableApplicationContext ac;
 
 	@Mock
 	private MessageHandler handlerA;
@@ -99,6 +100,12 @@ public void initialize() throws Exception {
 		failed = new AtomicBoolean(false);
 	}
 
+	@After
+	public void tearDown() {
+		this.executor.shutdownNow();
+		this.ac.close();
+	}
+
 	@Test
 	public void noFailoverNoLoadBalancing() {
 		DirectChannel channel = (DirectChannel) ac.getBean("noLoadBalancerNoFailover");

spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.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.
@@ -24,7 +24,6 @@
 
 import java.util.Comparator;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -38,6 +37,7 @@
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class PriorityChannelTests {
 
@@ -246,7 +246,7 @@ public void testTimeoutDoesNotElapse() throws InterruptedException {
 		final PriorityChannel channel = new PriorityChannel(1);
 		final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
 		final CountDownLatch latch = new CountDownLatch(1);
-		Executor executor = Executors.newSingleThreadScheduledExecutor();
+		ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
 		channel.send(new GenericMessage<String>("test-1"));
 		executor.execute(() -> {
 			sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 3000));
@@ -262,6 +262,7 @@ public void testTimeoutDoesNotElapse() throws InterruptedException {
 		Message<?> message2 = channel.receive();
 		assertNotNull(message2);
 		assertEquals("test-2", message2.getPayload());
+		executor.shutdownNow();
 	}
 
 	@Test