From fd9495577192858ac4c8299c5c5a139fd7d3ea42 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 18 Jan 2018 14:06:17 -0500 Subject: [PATCH] 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__ (perhaps just the gateway fix) --- .../integration/jms/JmsOutboundGateway.java | 11 +++++-- .../jms/JmsOutboundGatewayTests.java | 15 ++++++--- .../jms/JmsOutboundInsideChainTests.java | 5 +-- .../jms/OutboundGatewayConnectionTests.java | 9 ++++-- .../jms/OutboundGatewayFunctionTests.java | 31 ++++++++++++++----- .../jms/PollableJmsChannelTests.java | 11 ++++--- .../jms/SubscribableJmsChannelTests.java | 4 +-- .../jms/request_reply/PipelineJmsTests.java | 16 +++++++--- .../PipelineNamedReplyQueuesJmsTests.java | 16 +++++++--- ...eplyScenariosWithTempReplyQueuesTests.java | 7 +++-- 10 files changed, 87 insertions(+), 38 deletions(-) diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index faaa5f6745..1ac2292912 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.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. @@ -157,6 +157,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp private volatile long idleReplyContainerTimeout; + private volatile boolean wasStopped; + private ScheduledFuture idleTask; /** @@ -686,6 +688,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 { @@ -705,7 +711,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); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java index 83eace2cf9..6bb61f3c2f 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundGatewayTests.java +++ b/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; @@ -67,7 +68,7 @@ import org.springframework.util.ObjectUtils; * * @since 2.2.4 */ -public class JmsOutboundGatewayTests { +public class JmsOutboundGatewayTests extends ActiveMQMultiContextTests { private final Log logger = LogFactory.getLog(this.getClass()); @@ -100,8 +101,9 @@ public class JmsOutboundGatewayTests { gateway.setUseReplyContainer(true); ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties(); final List errors = new ArrayList(); + ExecutorService exec = Executors.newFixedThreadPool(10); ErrorHandlingTaskExecutor errorHandlingTaskExecutor = - new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), t -> { + new ErrorHandlingTaskExecutor(exec, t -> { errors.add(t); throw new RuntimeException(t); }); @@ -157,6 +159,7 @@ public class JmsOutboundGatewayTests { } finally { gateway.stop(); + exec.shutdownNow(); } } @@ -176,7 +179,8 @@ public class JmsOutboundGatewayTests { gateway.setReceiveTimeout(60000); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(() -> gateway.handleMessage(new GenericMessage("foo"))); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> gateway.handleMessage(new GenericMessage("foo"))); CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); @@ -197,6 +201,7 @@ public class JmsOutboundGatewayTests { gateway.stop(); connectionFactory1.destroy(); connectionFactory2.destroy(); + exec.shutdownNow(); } @Test @@ -216,7 +221,8 @@ public class JmsOutboundGatewayTests { gateway.setCorrelationKey("JMSCorrelationID"); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(() -> gateway.handleMessage(new GenericMessage("foo"))); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> gateway.handleMessage(new GenericMessage("foo"))); CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory( new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")); JmsTemplate template = new JmsTemplate(connectionFactory2); @@ -239,6 +245,7 @@ public class JmsOutboundGatewayTests { gateway.stop(); connectionFactory1.destroy(); connectionFactory2.destroy(); + exec.shutdownNow(); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java index 665342aba4..51d0882251 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsOutboundInsideChainTests.java +++ b/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; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java index bf60017dee..cd6dcf263a 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java +++ b/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; @@ -80,7 +81,8 @@ public class OutboundGatewayConnectionTests { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -107,7 +109,7 @@ public class OutboundGatewayConnectionTests { final CountDownLatch latch3 = new CountDownLatch(1); final CountDownLatch latch4 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + exec.execute(() -> { latch3.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -131,6 +133,7 @@ public class OutboundGatewayConnectionTests { broker.stop(); scheduler.destroy(); + exec.shutdownNow(); } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java index 409fe6b7b2..c6b1cf6ecc 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java +++ b/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; @@ -56,7 +57,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; * @since 2.2 * */ -public class OutboundGatewayFunctionTests { +public class OutboundGatewayFunctionTests extends ActiveMQMultiContextTests { private static Destination requestQueue1 = new ActiveMQQueue("request1"); @@ -102,7 +103,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -124,6 +126,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdown(); } @Test @@ -145,7 +148,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -170,6 +174,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -192,7 +197,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); @@ -214,6 +220,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -235,7 +242,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference<>(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); @@ -260,6 +268,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -282,7 +291,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference<>(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); @@ -304,6 +314,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -324,7 +335,8 @@ public class OutboundGatewayFunctionTests { final AtomicReference reply = new AtomicReference<>(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { latch1.countDown(); try { reply.set(gateway.handleRequestMessage(new GenericMessage<>("foo"))); @@ -349,6 +361,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); scheduler.destroy(); + exec.shutdownNow(); } @Test @@ -371,7 +384,8 @@ public class OutboundGatewayFunctionTests { gateway.setReceiveTimeout(20000); gateway.afterPropertiesSet(); gateway.start(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { JmsTemplate template = new JmsTemplate(); template.setConnectionFactory(getConnectionFactory()); template.setReceiveTimeout(20000); @@ -393,6 +407,7 @@ public class OutboundGatewayFunctionTests { gateway.stop(); assertFalse(container.isRunning()); scheduler.destroy(); + exec.shutdownNow(); } private void receiveAndSend(JmsTemplate template) { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java index 9e65970f17..bf629f6594 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/PollableJmsChannelTests.java +++ b/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,6 +28,7 @@ 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; @@ -59,7 +60,7 @@ import org.springframework.messaging.support.GenericMessage; * @author Gunnar Hillert * @author Artem Bilan */ -public class PollableJmsChannelTests { +public class PollableJmsChannelTests extends ActiveMQMultiContextTests { private ActiveMQConnectionFactory connectionFactory; @@ -188,7 +189,8 @@ public class PollableJmsChannelTests { assertTrue(sent1); final AtomicReference message = new AtomicReference(); final CountDownLatch latch1 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { message.set(receiver.receive(queue)); latch1.countDown(); }); @@ -201,7 +203,7 @@ public class PollableJmsChannelTests { final CountDownLatch latch2 = new CountDownLatch(1); boolean sent2 = channel.send(MessageBuilder.withPayload("foo").setPriority(6).build()); assertTrue(sent2); - Executors.newSingleThreadExecutor().execute(() -> { + exec.execute(() -> { message.set(receiver.receive(queue)); latch2.countDown(); }); @@ -210,6 +212,7 @@ public class PollableJmsChannelTests { assertEquals(6, message.get().getJMSPriority()); assertTrue(message.get().getJMSExpiration() <= System.currentTimeMillis() + ttl); assertTrue(message.get().toString().contains("persistent = false")); + exec.shutdownNow(); } @Test diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java index 72c31a42b4..9f6e6c28ec 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.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. @@ -68,7 +68,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; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java index a44ccdd7c1..f6e6d394f5 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java +++ b/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; @@ -46,7 +47,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 +59,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 -> @@ -186,7 +192,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(); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java index 52119ec0b7..9a661b57ea 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java +++ b/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; @@ -51,7 +52,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 +64,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 -> @@ -194,7 +200,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(); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java index fdb8f4b597..4da7cf00bc 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java +++ b/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. @@ -23,7 +23,7 @@ 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; @@ -210,7 +210,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(); @@ -250,6 +250,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) {