Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4379: JMS OG Shutdown reply container on stop

Browse Source 
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)
This commit is contained in:
Gary Russell
2018-01-18 14:06:17 -05:00
parent 7e263aa618
commit fd94955771
10 changed files with 87 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);

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

@@ -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<Throwable> errors = new ArrayList<Throwable>();
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<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);
@@ -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<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);
@@ -239,6 +245,7 @@ public class JmsOutboundGatewayTests {
gateway.stop();
connectionFactory1.destroy();
connectionFactory2.destroy();
exec.shutdownNow();
}
}

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

@@ -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;

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

@@ -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<Object> reply = new AtomicReference<Object>();
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<String>("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<String>("foo")));
@@ -131,6 +133,7 @@ public class OutboundGatewayConnectionTests {
broker.stop();
scheduler.destroy();
exec.shutdownNow();
}
}

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

@@ -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<Object> reply = new AtomicReference<Object>();
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<String>("foo")));
@@ -124,6 +126,7 @@ public class OutboundGatewayFunctionTests {
gateway.stop();
scheduler.destroy();
exec.shutdown();
}
@Test
@@ -145,7 +148,8 @@ public class OutboundGatewayFunctionTests {
final AtomicReference<Object> reply = new AtomicReference<Object>();
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<String>("foo")));
@@ -170,6 +174,7 @@ public class OutboundGatewayFunctionTests {
gateway.stop();
scheduler.destroy();
exec.shutdownNow();
}
@Test
@@ -192,7 +197,8 @@ public class OutboundGatewayFunctionTests {
final AtomicReference<Object> reply = new AtomicReference<Object>();
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<String>("foo")));
@@ -214,6 +220,7 @@ public class OutboundGatewayFunctionTests {
gateway.stop();
scheduler.destroy();
exec.shutdownNow();
}
@Test
@@ -235,7 +242,8 @@ public class OutboundGatewayFunctionTests {
final AtomicReference<Object> 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<Object> 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<Object> 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) {

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

@@ -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<javax.jms.Message> message = new AtomicReference<javax.jms.Message>();
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

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

@@ -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;

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

@@ -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();

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

@@ -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();

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

@@ -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) {