From f1bd6e3bac07e894a4d0910a58a79ddc5b3015d3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 12 Nov 2015 11:47:10 -0500 Subject: [PATCH] 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 --- .../integration/jms/JmsOutboundGateway.java | 12 +++++- .../PipelineNamedReplyQueuesJmsTests.java | 39 +++++++++++++++---- .../pipeline-named-queue-03a.xml | 30 ++++++++++++-- .../request_reply/pipeline-named-queue-04.xml | 24 ++++++++++-- .../src/test/resources/log4j.properties | 6 +-- 5 files changed, 92 insertions(+), 19 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 00122d63f5..5030bf44df 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 @@ -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> replies = - new HashMap>(); + new ConcurrentHashMap>(); private final ConcurrentHashMap earlyOrLateReplies = new ConcurrentHashMap(); @@ -1178,6 +1180,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp LinkedBlockingQueue 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) { 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 b5cd5cdbfa..f2760f6980 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 @@ -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(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 messages = new LinkedBlockingQueue(); + + public Message capture(Message message) { + messages.add("\n[" + Thread.currentThread().getName() + "] " + message); + return message; + } + + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml index aa0d9ed4d4..5990be1c34 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml @@ -10,6 +10,7 @@ + + + + + + + + @@ -33,6 +42,7 @@ + + + + + - + + + + - + + + + - + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml index 0c2fbd8076..deb9a50c6c 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml @@ -10,14 +10,22 @@ + + + + + + + + @@ -34,20 +43,29 @@ + + + + + - + + + + diff --git a/spring-integration-jms/src/test/resources/log4j.properties b/spring-integration-jms/src/test/resources/log4j.properties index b8ebc43ebe..4eb3e380a2 100644 --- a/spring-integration-jms/src/test/resources/log4j.properties +++ b/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.jdbc=DEBUG -log4j.category.org.springframework.jms=ERROR +log4j.category.org.springframework.integration=ERROR +log4j.category.org.springframework.integration.jms.request_reply=INFO +log4j.category.si.jmsgateway.debug=DEBUG