INT-3885: Fix JMS Outbound Gateway Concurrency

JIRA: https://jira.spring.io/browse/INT-3885

Possible dropped reply, causing timeout.

WARN org.springframework.integration.jms.JmsOutboundGateway#1.replyListener-1 jms.JmsOutboundGateway:1202
    - Failed to consume reply with correlationId 164a49bf-c41d-4c0b-b012-55deecf001d1_2
      java.lang.RuntimeException: No sender waiting for reply

- Reproduced by running the test in a loop
- Cleaned up test to aid debugging - capture a unique message at each stage
- Added additional debug logging to th gateway

Polishing

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

@@ -17,7 +17,6 @@
 package org.springframework.integration.jms;
 
 import java.util.Date;
-import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -42,6 +41,9 @@ import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.context.Lifecycle;
 import org.springframework.expression.Expression;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -134,7 +136,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 	private final String gatewayCorrelation = UUID.randomUUID().toString();
 
 	private final Map<String, LinkedBlockingQueue<javax.jms.Message>> replies =
-			new HashMap<String, LinkedBlockingQueue<javax.jms.Message>>();
+			new ConcurrentHashMap<String, LinkedBlockingQueue<javax.jms.Message>>();
 
 	private final ConcurrentHashMap<String, TimedReply> earlyOrLateReplies =
 			new ConcurrentHashMap<String, JmsOutboundGateway.TimedReply>();
@@ -1178,6 +1180,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 			LinkedBlockingQueue<javax.jms.Message> queue = this.replies.get(correlationId);
 			if (queue == null) {
 				if (this.correlationKey != null) {
+					Log debugLogger = LogFactory.getLog("si.jmsgateway.debug");
+					if (debugLogger.isDebugEnabled()) {
+						Object siMessage = this.messageConverter.fromMessage(message);
+						debugLogger.debug("No pending reply for " + siMessage + " with correlationId: "
+								+ correlationId + " pending replies: " + this.replies.keySet());
+					}
 					throw new RuntimeException("No sender waiting for reply");
 				}
 				synchronized (this.earlyOrLateReplies) {

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

@@ -18,9 +18,11 @@ package org.springframework.integration.jms.request_reply;
 import static org.junit.Assert.assertEquals;
 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.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -37,8 +39,9 @@ import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.gateway.RequestReplyExchanger;
 import org.springframework.integration.jms.ActiveMQMultiContextTests;
 import org.springframework.integration.jms.config.ActiveMqTestUtils;
-import org.springframework.messaging.support.GenericMessage;
 import org.springframework.integration.test.support.LongRunningIntegrationTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
 /**
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -113,7 +116,7 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 	 */
 	@Test
 	public void testPipeline3a() throws Exception{
-		int timeouts = this.test("pipeline-named-queue-03a.xml", 20000);
+		int timeouts = this.test("pipeline-named-queue-03a.xml", 50000);
 		assertEquals(0, timeouts);
 	}
 
@@ -123,7 +126,8 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 	 */
 	@Test
 	public void testPipeline4() throws Exception{
-		this.test("pipeline-named-queue-04.xml");
+		int timeouts = this.test("pipeline-named-queue-04.xml", 30000);
+		assertEquals(0, timeouts);
 	}
 
 	/**
@@ -167,19 +171,23 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 			final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
 			final CountDownLatch latch = new CountDownLatch(requests);
 
-			for (int i = 0; i < requests; i++) {
+			for (int i = 1000000; i < 1000000 + requests * 100000; i += 100000) {
 				final int y = i;
 				executor.execute(new Runnable() {
+					@Override
 					public void run() {
 						try {
 							assertEquals(y + offset, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
 							successCounter.incrementAndGet();
-						} catch (MessageTimeoutException e) {
+						}
+						catch (MessageTimeoutException e) {
 							timeoutCounter.incrementAndGet();
-						} catch (Throwable t) {
+						}
+						catch (Throwable t) {
 							t.printStackTrace();
 							failureCounter.incrementAndGet();
-						} finally {
+						}
+						finally {
 							latch.countDown();
 						}
 					}
@@ -198,7 +206,22 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
 			logger.info("Success: " + successCounter.get());
 			logger.info("Timeout: " + timeoutCounter.get());
 			logger.info("Failure: " + failureCounter.get());
-			context.destroy();
+			if (timeoutCounter.get() > 0 && context.containsBean("capture")) {
+				logger.info(context.getBean(Capture.class).messages);
+			}
+			context.close();
 		}
 	}
+
+	public static class Capture {
+
+		private final BlockingQueue<String> messages = new LinkedBlockingQueue<String>();
+
+		public Message<?> capture(Message<?> message) {
+			messages.add("\n[" + Thread.currentThread().getName() + "] " + message);
+			return message;
+		}
+
+	}
+
 }

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml

@@ -10,6 +10,7 @@
 	<int:gateway default-request-channel="pipeline03" default-request-timeout="10000" default-reply-timeout="10000"/>
 
 	<int-jms:outbound-gateway request-channel="pipeline03"
+		reply-channel="reply1"
 		connection-factory="connectionFactory"
 		reply-destination-name="pipeline03a-01"
 		correlation-key="JMSCorrelationID"
@@ -18,6 +19,13 @@
 		<int-jms:reply-listener />
 	</int-jms:outbound-gateway>
 
+	<bean id="capture" class="org.springframework.integration.jms.request_reply.PipelineNamedReplyQueuesJmsTests.Capture" />
+
+	<int:chain input-channel="reply1">
+		<int:transformer expression="payload + 10000"/>
+		<int:service-activator ref="capture" />
+	</int:chain>
+
 	<int-jms:inbound-gateway request-channel="jmsIn"
 		request-destination-name="siOutQueue03a"
 		connection-factory="connectionFactory"
@@ -25,6 +33,7 @@
 		reply-timeout="10000"/>
 
 	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:service-activator ref="capture" />
 		<int:header-enricher>
 			<int:header name="delay" expression="0"/>
 		</int:header-enricher>
@@ -33,6 +42,7 @@
 	</int:chain>
 
 	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+		reply-channel="reply2"
 		connection-factory="connectionFactory"
 		reply-destination-name="pipeline03a-01"
 		correlation-key="corr03a"
@@ -41,6 +51,11 @@
 		<int-jms:reply-listener />
 	</int-jms:outbound-gateway>
 
+	<int:chain input-channel="reply2">
+		<int:transformer expression="payload + 10000"/>
+		<int:service-activator ref="capture" />
+	</int:chain>
+
 	<int-jms:inbound-gateway request-channel="anotherIn"
 		request-destination-name="anotherGatewayQueue03a"
 		correlation-key="corr03a"
@@ -48,7 +63,10 @@
 		concurrent-consumers="10"
 		reply-timeout="10000"/>
 
-	<int:transformer input-channel="anotherIn" expression="payload" output-channel="toThird" />
+	<int:chain input-channel="anotherIn" output-channel="toThird">
+		<int:transformer expression="payload + 10000" />
+		<int:service-activator ref="capture" />
+	</int:chain>
 
 	<int-jms:outbound-gateway request-channel="toThird"
 		reply-channel="add10kOnTheWayBack"
@@ -60,7 +78,10 @@
 		<int-jms:reply-listener />
 	</int-jms:outbound-gateway>
 
-	<int:transformer input-channel="add10kOnTheWayBack" expression="payload + 10000" />
+	<int:chain input-channel="add10kOnTheWayBack">
+		<int:transformer expression="payload + 10000" />
+		<int:service-activator ref="capture" />
+	</int:chain>
 
 	<int-jms:inbound-gateway request-channel="thirdIn"
 		request-destination-name="thirdGatewayQueue03a"
@@ -69,7 +90,10 @@
 		concurrent-consumers="10"
 		reply-timeout="10000"/>
 
-	<int:transformer input-channel="thirdIn" expression="payload + 10000" />
+	<int:chain input-channel="thirdIn">
+		<int:transformer expression="payload + 10000" />
+		<int:service-activator ref="capture" />
+	</int:chain>
 
 	<bean id="connectionFactory"
 		class="org.springframework.jms.connection.CachingConnectionFactory">

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml

@@ -10,14 +10,22 @@
 	<int:gateway default-request-channel="pipeline04" default-request-timeout="10000" default-reply-timeout="10000"/>
 
 	<int-jms:outbound-gateway request-channel="pipeline04"
+		reply-channel="reply1"
 		connection-factory="connectionFactory"
 		reply-destination-name="pipeline04-01"
 		correlation-key="foo"
-		receive-timeout="10000"
+		receive-timeout="30000"
 		request-destination-name="siOutQueue04">
 		<int-jms:reply-listener />
 	</int-jms:outbound-gateway>
 
+	<bean id="capture" class="org.springframework.integration.jms.request_reply.PipelineNamedReplyQueuesJmsTests.Capture" />
+
+	<int:chain input-channel="reply1">
+		<int:transformer expression="payload + 10000"/>
+		<int:service-activator ref="capture" />
+	</int:chain>
+
 	<int-jms:inbound-gateway request-channel="jmsIn"
 		request-destination-name="siOutQueue04"
 		connection-factory="connectionFactory"
@@ -26,6 +34,7 @@
 		reply-timeout="10000"/>
 
 	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:service-activator ref="capture" />
 		<int:header-enricher>
 			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
 		</int:header-enricher>
@@ -34,20 +43,29 @@
 	</int:chain>
 
 	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+		reply-channel="reply2"
 		connection-factory="connectionFactory"
 		reply-destination-name="pipeline04-02"
-		receive-timeout="10000"
+		receive-timeout="30000"
 		request-destination-name="anotherGatewayQueue04">
 		<int-jms:reply-listener />
 	</int-jms:outbound-gateway>
 
+	<int:chain input-channel="reply2">
+		<int:transformer expression="payload + 10000"/>
+		<int:service-activator ref="capture" />
+	</int:chain>
+
 	<int-jms:inbound-gateway request-channel="anotherIn"
 		request-destination-name="anotherGatewayQueue04"
 		connection-factory="connectionFactory"
 		concurrent-consumers="10"
 		reply-timeout="10000"/>
 
-	<int:transformer input-channel="anotherIn" expression="payload"/>
+	<int:chain input-channel="anotherIn">
+		<int:transformer expression="payload + 10000" />
+		<int:service-activator ref="capture" />
+	</int:chain>
 
 	<bean id="connectionFactory"
 		class="org.springframework.jms.connection.CachingConnectionFactory">

spring-integration-jms/src/test/resources/log4j.properties

@@ -6,6 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m
 
 
 log4j.category.org.springframework=ERROR
-#log4j.category.org.springframework.integration.jms=DEBUG
+log4j.category.org.springframework.integration=ERROR
-# log4j.category.org.springframework.integration.jdbc=DEBUG
+log4j.category.org.springframework.integration.jms.request_reply=INFO
-log4j.category.org.springframework.jms=ERROR
+log4j.category.si.jmsgateway.debug=DEBUG