From 8aa91d1db0d6c99c75bae771f2519f8d3e85ea9e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 18 Jan 2018 17:37:17 -0500 Subject: [PATCH] Shutdown test executors in -core - don't use `ExecutorService` as `@Bean` - spring can't stop them - also add log adjuster for ftp test that sometimes fails --- ...bstractCorrelatingMessageHandlerTests.java | 6 +++-- .../aggregator/AggregatorTests.java | 2 +- .../BarrierMessageHandlerTests.java | 16 +++++++++----- .../CorrelatingMessageBarrierTests.java | 13 ++++++----- .../CorrelatingMessageHandlerTests.java | 7 ++++-- ...regatorWithCustomReleaseStrategyTests.java | 2 +- .../channel/CGLibProxyChannelTests.java | 10 ++++----- .../channel/ExecutorChannelTests.java | 21 ++++++++++++------ ...dDispatcherConfigurationScenarioTests.java | 13 ++++++++--- .../channel/PriorityChannelTests.java | 7 +++--- .../channel/QueueChannelTests.java | 7 +++--- .../RoundRobinDispatcherConcurrentTests.java | 22 ++++++++++++------- .../correlation/CorrelationHandlerTests.java | 8 ++++--- .../dsl/flows/IntegrationFlowTests.java | 14 +++++++++--- .../reactivestreams/ReactiveStreamsTests.java | 8 +++++-- .../gateway/GatewayProxyFactoryBeanTests.java | 7 +++--- .../handler/AsyncHandlerTests.java | 12 +++++++++- .../advice/AdvisedMessageHandlerTests.java | 7 ++++-- .../store/SimpleMessageStoreTests.java | 11 +++++++--- .../FtpStreamingMessageSourceTests.java | 6 +++++ 20 files changed, 137 insertions(+), 62 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index dd9f6d805e..1fc8dd8b68 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/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 @@ public class AbstractCorrelatingMessageHandlerTests { /* * 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 @@ public class AbstractCorrelatingMessageHandlerTests { assertEquals(1, ((MessageGroup) outputMessages.get(1).getPayload()).size()); // 'qux' assertNull(discards.receive(0)); + exec.shutdownNow(); } @Test // INT-2833 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index a02e193483..0334fcf62d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -272,7 +272,7 @@ public class AggregatorTests { 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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java index 57744e6a93..9100003dac 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/BarrierMessageHandlerTests.java +++ b/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 class BarrierMessageHandlerTests { assertEquals("bar", result.get(1)); assertEquals(0, suspensions.size()); assertEquals(0, inProcess.size()); + exec.shutdownNow(); } @Test @@ -141,8 +142,8 @@ public class BarrierMessageHandlerTests { 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 class BarrierMessageHandlerTests { assertEquals("foo", result.get(0)); assertEquals("bar", result.get(1)); assertEquals(0, suspensions.size()); + exec.shutdownNow(); } @Test @@ -169,7 +171,8 @@ public class BarrierMessageHandlerTests { 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 class BarrierMessageHandlerTests { assertSame(discard, triggerMessage); handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build()); assertEquals(0, suspensions.size()); + exec.shutdownNow(); } @Test @@ -218,7 +222,8 @@ public class BarrierMessageHandlerTests { handler.afterPropertiesSet(); final AtomicReference exception = new AtomicReference(); 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 class BarrierMessageHandlerTests { assertTrue(latch.await(10, TimeUnit.SECONDS)); assertSame(exc, exception.get().getCause()); assertEquals(0, suspensions.size()); + exec.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java index 1b2745fb20..f28e85df17 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java +++ b/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 static org.mockito.Mockito.when; 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 class CorrelatingMessageBarrierTests { 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 class CorrelatingMessageBarrierTests { trackingReleaseStrategy.release("foo"); assertThat((barrier.receive()), is(notNullValue())); } + exec.shutdownNow(); } - private void sendAsynchronously(final MessageHandler handler, final Message message, final CountDownLatch start, final CountDownLatch sent) { - Executors.newSingleThreadExecutor().execute(() -> { + private void sendAsynchronously(final MessageHandler handler, final Message message, + final CountDownLatch start, final CountDownLatch sent, ExecutorService exec) { + exec.execute(() -> { try { start.await(); } @@ -119,7 +123,6 @@ public class CorrelatingMessageBarrierTests { handler.handleMessage(message); sent.countDown(); }); - } private Message testMessage() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index d1bffab1c5..168e64143a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/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 static org.mockito.Mockito.when; 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 class CorrelatingMessageHandlerTests { 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 class CorrelatingMessageHandlerTests { assertTrue(bothMessagesHandled.await(10, TimeUnit.SECONDS)); assertEquals(0, store.expireMessageGroups(10000)); + exec.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java index a009c2bd99..b19bad68c2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java @@ -92,7 +92,7 @@ public class AggregatorWithCustomReleaseStrategyTests { }); } - 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; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java index ea426a7941..b1a627a09e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java +++ b/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.assertNotNull; 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.context.annotation.Configuration; 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 class CGLibProxyChannelTests { } @Bean - public ExecutorService executor() { - return Executors.newCachedThreadPool(); + public Executor executor() { + return new ThreadPoolTaskExecutor(); } private ProxyFactoryBean createProxyFactory(MessageChannel channel) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java index 0b30d0b865..1635cd56f8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java +++ b/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 static org.mockito.Mockito.verify; 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 class ExecutorChannelTests { @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 class ExecutorChannelTests { 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 class ExecutorChannelTests { 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 class ExecutorChannelTests { assertEquals(0, handler1.count.get()); assertEquals(0, handler3.count.get()); assertEquals(numberOfMessages, handler2.count.get()); + exec.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java index 8f07a27654..02bc0d4e6c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java +++ b/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.ExecutorService; 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.Mock; 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 exceptionRegistry; - private ApplicationContext ac; + private ConfigurableApplicationContext ac; @Mock private MessageHandler handlerA; @@ -99,6 +100,12 @@ public class MixedDispatcherConfigurationScenarioTests { failed = new AtomicBoolean(false); } + @After + public void tearDown() { + this.executor.shutdownNow(); + this.ac.close(); + } + @Test public void noFailoverNoLoadBalancing() { DirectChannel channel = (DirectChannel) ac.getBean("noLoadBalancerNoFailover"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java index 60ae7f9f62..dcecf3d752 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/PriorityChannelTests.java +++ b/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 static org.junit.Assert.assertTrue; 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 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher + * @author Gary Russell */ public class PriorityChannelTests { @@ -246,7 +246,7 @@ public class PriorityChannelTests { 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("test-1")); executor.execute(() -> { sentSecondMessage.set(channel.send(new GenericMessage("test-2"), 3000)); @@ -262,6 +262,7 @@ public class PriorityChannelTests { Message message2 = channel.receive(); assertNotNull(message2); assertEquals("test-2", message2.getPayload()); + executor.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index 80c9c1e033..55def0220c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.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. @@ -23,7 +23,7 @@ import static org.junit.Assert.assertTrue; import java.util.List; 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; import java.util.concurrent.atomic.AtomicBoolean; @@ -68,7 +68,7 @@ public class QueueChannelTests { final QueueChannel channel = new QueueChannel(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executor singleThreadExecutor = Executors.newSingleThreadExecutor(); + ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor(); Runnable receiveTask1 = () -> { Message message = channel.receive(0); if (message != null) { @@ -91,6 +91,7 @@ public class QueueChannelTests { singleThreadExecutor.execute(receiveTask2); latch2.await(); assertTrue(messageReceived.get()); + singleThreadExecutor.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java index 38296ae71e..886715fb98 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/RoundRobinDispatcherConcurrentTests.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. @@ -26,6 +26,7 @@ import java.util.concurrent.CountDownLatch; 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; @@ -49,7 +50,7 @@ public class RoundRobinDispatcherConcurrentTests { private final UnicastingDispatcher dispatcher = new UnicastingDispatcher(); - private final ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor(); + private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); @Mock private MessageHandler handler1; @@ -69,9 +70,14 @@ public class RoundRobinDispatcherConcurrentTests { @Before public void initialize() throws Exception { dispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); - scheduler.setCorePoolSize(10); - scheduler.setMaxPoolSize(10); - scheduler.initialize(); + executor.setCorePoolSize(10); + executor.setMaxPoolSize(10); + executor.initialize(); + } + + @After + public void tearDown() { + this.executor.shutdown(); } @Test(timeout = 1000) @@ -97,7 +103,7 @@ public class RoundRobinDispatcherConcurrentTests { allDone.countDown(); }; for (int i = 0; i < TOTAL_EXECUTIONS; i++) { - scheduler.execute(messageSenderTask); + executor.execute(messageSenderTask); } start.countDown(); allDone.await(); @@ -131,7 +137,7 @@ public class RoundRobinDispatcherConcurrentTests { allDone.countDown(); }; for (int i = 0; i < TOTAL_EXECUTIONS; i++) { - scheduler.execute(messageSenderTask); + executor.execute(messageSenderTask); } start.countDown(); allDone.await(); @@ -164,7 +170,7 @@ public class RoundRobinDispatcherConcurrentTests { } }; for (int i = 0; i < TOTAL_EXECUTIONS; i++) { - scheduler.execute(messageSenderTask); + executor.execute(messageSenderTask); } start.countDown(); allDone.await(5000, TimeUnit.MILLISECONDS); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index 35423427a5..d54f376eb6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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,7 +27,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.Executor; -import java.util.concurrent.Executors; import java.util.stream.Collectors; import org.junit.Test; @@ -51,6 +50,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; 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; @@ -157,7 +157,9 @@ public class CorrelationHandlerTests { @Bean public Executor taskExecutor() { - return Executors.newCachedThreadPool(); + ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor(); + tpte.setCorePoolSize(50); + return tpte; } @Bean diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index f66d211568..9d8cf43fa9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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,7 +28,7 @@ import static org.junit.Assert.fail; import java.io.Serializable; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -86,6 +86,7 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.stereotype.Component; import org.springframework.stereotype.Service; @@ -614,7 +615,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow subscribersFlow() { return flow -> flow - .publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s + .publishSubscribeChannel(executor(), s -> s .subscribe(f -> f .handle((p, h) -> p / 2) .channel(MessageChannels.queue("subscriber1Results"))) @@ -625,6 +626,13 @@ public class IntegrationFlowTests { .channel(MessageChannels.queue("subscriber3Results")); } + @Bean + public Executor executor() { + ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor(); + tpte.setCorePoolSize(50); + return tpte; + } + @Bean public IntegrationFlow wireTapFlow1() { return IntegrationFlows.from("tappedChannel1") diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index 9b10aa5cff..df02a3ab00 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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.Arrays; import java.util.Date; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -60,6 +61,7 @@ import reactor.core.publisher.Flux; /** * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -116,8 +118,9 @@ public class ReactiveStreamsTests { .doOnNext(p -> latch.countDown()) .subscribe(); + ExecutorService exec = Executors.newSingleThreadExecutor(); Future> future = - Executors.newSingleThreadExecutor().submit(() -> + exec.submit(() -> Flux.just("11,12,13") .map(v -> v.split(",")) .flatMapIterable(Arrays::asList) @@ -138,6 +141,7 @@ public class ReactiveStreamsTests { assertNotNull(integers); assertEquals(7, integers.size()); + exec.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java index 555646d6bc..7df66ee349 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.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. @@ -31,7 +31,7 @@ import java.util.Collections; import java.util.Map; 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.TimeUnit; @@ -214,7 +214,7 @@ public class GatewayProxyFactoryBeanTests { final TestService service = (TestService) context.getBean("proxy"); final String[] results = new String[numRequests]; final CountDownLatch latch = new CountDownLatch(numRequests); - Executor executor = Executors.newFixedThreadPool(numRequests); + ExecutorService executor = Executors.newFixedThreadPool(numRequests); for (int i = 0; i < numRequests; i++) { final int count = i; executor.execute(() -> { @@ -237,6 +237,7 @@ public class GatewayProxyFactoryBeanTests { assertEquals(numRequests, interceptor.getSentCount()); assertEquals(numRequests, interceptor.getReceivedCount()); context.close(); + executor.shutdownNow(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java index f309e9b6d1..ef2581d3a3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java @@ -32,10 +32,12 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; +import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -75,14 +77,17 @@ public class AsyncHandlerTests { private volatile CountDownLatch exceptionLatch = new CountDownLatch(1); + private ExecutorService executor; + @Before public void setup() { + this.executor = Executors.newSingleThreadExecutor(); this.handler = new AbstractReplyProducingMessageHandler() { @Override protected Object handleRequestMessage(Message requestMessage) { final SettableListenableFuture future = new SettableListenableFuture(); - Executors.newSingleThreadExecutor().execute(() -> { + AsyncHandlerTests.this.executor.execute(() -> { try { latch.await(10, TimeUnit.SECONDS); switch (whichTest) { @@ -118,6 +123,11 @@ public class AsyncHandlerTests { }).when(logger).error(anyString(), any(Throwable.class)); } + @After + public void tearDown() { + this.executor.shutdownNow(); + } + @Test public void testGoodResult() { this.whichTest = 0; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java index 89be7ab505..d4b5a9ba34 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/AdvisedMessageHandlerTests.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. @@ -39,6 +39,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -890,7 +891,8 @@ public class AdvisedMessageHandlerTests { PollableChannel inputChannel = new QueueChannel(); PollingConsumer consumer = new PollingConsumer(inputChannel, message -> { }); consumer.setAdviceChain(Collections.singletonList(advice)); - consumer.setTaskExecutor(new ErrorHandlingTaskExecutor(Executors.newSingleThreadExecutor(), t -> { })); + ExecutorService exec = Executors.newSingleThreadExecutor(); + consumer.setTaskExecutor(new ErrorHandlingTaskExecutor(exec, t -> { })); consumer.setBeanFactory(mock(BeanFactory.class)); consumer.afterPropertiesSet(); consumer.setTaskScheduler(mock(TaskScheduler.class)); @@ -916,6 +918,7 @@ public class AdvisedMessageHandlerTests { "an attempt to advise method 'call' in " + "'org.springframework.integration.endpoint.AbstractPollingEndpoint")); consumer.stop(); + exec.shutdownNow(); } public void filterDiscardNoAdvice() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 9fa3b2048f..1b932c7eb1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.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. @@ -30,6 +30,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -107,7 +108,8 @@ public class SimpleMessageStoreTests { final CountDownLatch message2Latch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { store2.addMessage(testMessage2); message2Latch.countDown(); }); @@ -119,6 +121,7 @@ public class SimpleMessageStoreTests { assertTrue(message2Latch.await(10, TimeUnit.SECONDS)); Message t2 = store2.getMessage(testMessage2.getHeaders().getId()); assertEquals(testMessage2, t2); + exec.shutdownNow(); } @Test(expected = MessagingException.class) @@ -150,7 +153,8 @@ public class SimpleMessageStoreTests { final CountDownLatch message2Latch = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { store2.addMessageToGroup("foo", testMessage2); message2Latch.countDown(); }); @@ -161,6 +165,7 @@ public class SimpleMessageStoreTests { assertTrue(message2Latch.await(10, TimeUnit.SECONDS)); MessageGroup messageGroup = store2.getMessageGroup("foo"); messageGroup.getMessages().contains(testMessage2); + exec.shutdownNow(); } @Test(expected = MessagingException.class) diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index 5908898da1..16e6b3646b 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -26,6 +26,7 @@ import java.io.InputStream; import java.util.Comparator; import org.apache.commons.net.ftp.FTPFile; +import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; @@ -46,6 +47,7 @@ import org.springframework.integration.ftp.FtpTestSupport; import org.springframework.integration.ftp.session.FtpFileInfo; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.test.rule.Log4j2LevelAdjuster; import org.springframework.integration.transformer.StreamTransformer; import org.springframework.messaging.Message; import org.springframework.scheduling.support.PeriodicTrigger; @@ -71,6 +73,10 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport { @Autowired private SourcePollingChannelAdapter adapter; + @Rule + public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.debug() + .categories("org.springframework.integration", "org.apache.commons"); + @SuppressWarnings("unchecked") @Test public void testAllContents() {