From 1cc1b2ef79cd5742c65bc5163261e7b4db80ccb2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 20 Nov 2015 12:25:32 -0500 Subject: [PATCH] INT-3895: Fix `MonitorTests` Race Condition JIRA: https://jira.spring.io/browse/INT-3895 * `@Ignore` `DelayerUsageTests.testDelayWithCustomScheduler()` as very weak test. (We can consider it to remove at all: doesn't test anything from our side) * Add `LogAdjustingTestSupport` diagnostic to the `OutboundGatewayFunctionTests` Tentative commit to see more logs from Travis Additional diagnostics More STOMP Diagnostics --- .../config/xml/DelayerUsageTests.java | 32 ++++++---- .../jms/OutboundGatewayFunctionTests.java | 3 +- .../integration/monitor/MonitorTests.java | 18 +++--- .../stomp/AbstractStompSessionManager.java | 58 +++++++++++++++++-- .../client/StompServerIntegrationTests.java | 3 +- 5 files changed, 87 insertions(+), 27 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java index d6b03e2df8..bf97bb62d2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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,15 +20,17 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.integration.support.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -41,19 +43,23 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration public class DelayerUsageTests { - @Autowired @Qualifier("inputA") + @Autowired + @Qualifier("inputA") private MessageChannel inputA; @Autowired private MessageChannel delayerInsideChain; - @Autowired @Qualifier("outputA") + @Autowired + @Qualifier("outputA") private PollableChannel outputA; - @Autowired @Qualifier("inputB") + @Autowired + @Qualifier("inputB") private MessageChannel inputB; - @Autowired @Qualifier("outputB1") + @Autowired + @Qualifier("outputB1") private PollableChannel outputB1; @Autowired @@ -64,7 +70,7 @@ public class DelayerUsageTests { @Test - public void testDelayWithDefaultScheduler(){ + public void testDelayWithDefaultScheduler() { long start = System.currentTimeMillis(); inputA.send(new GenericMessage("Hello")); assertNotNull(outputA.receive(10000)); @@ -72,7 +78,7 @@ public class DelayerUsageTests { } @Test - public void testDelayWithDefaultSchedulerCustomDelayHeader(){ + public void testDelayWithDefaultSchedulerCustomDelayHeader() { MessageBuilder builder = MessageBuilder.withPayload("Hello"); // set custom delay header builder.setHeader("foo", 2000); @@ -83,7 +89,8 @@ public class DelayerUsageTests { } @Test - public void testDelayWithCustomScheduler(){ + @Ignore("Enough wonky test based on the timeout and hardware") + public void testDelayWithCustomScheduler() { long start = System.currentTimeMillis(); inputB.send(new GenericMessage("1")); inputB.send(new GenericMessage("2")); @@ -107,7 +114,7 @@ public class DelayerUsageTests { } @Test //INT-1132 - public void testDelayerInsideChain(){ + public void testDelayerInsideChain() { long start = System.currentTimeMillis(); delayerInsideChain.send(new GenericMessage("Hello")); Message message = outputA.receive(10000); @@ -126,10 +133,13 @@ public class DelayerUsageTests { assertEquals("test", message.getPayload()); } - public static class SampleService{ + public static class SampleService { + public String processMessage(String message) throws Exception { Thread.sleep(500); return message; } + } + } 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 8406fdff2d..053270066d 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 @@ -40,6 +40,7 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.test.support.LogAdjustingTestSupport; import org.springframework.integration.test.util.TestUtils; import org.springframework.jms.JmsException; import org.springframework.jms.connection.CachingConnectionFactory; @@ -55,7 +56,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; * @since 2.2 * */ -public class OutboundGatewayFunctionTests { +public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport { private static Destination requestQueue1 = new ActiveMQQueue("request1"); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests.java index 9fc0d5327d..873faeb10b 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MonitorTests.java @@ -42,11 +42,13 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractMessageSource; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.support.management.DefaultMessageChannelMetrics; import org.springframework.integration.support.management.DefaultMessageHandlerMetrics; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.annotation.Repeat; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -85,30 +87,30 @@ public class MonitorTests { @Test public void testStats() throws InterruptedException { - final CountDownLatch afterHandleLatch = new CountDownLatch(1); + final CountDownLatch afterSendLatch = new CountDownLatch(1); - DefaultMessageHandlerMetrics handlerMetrics = - TestUtils.getPropertyValue(this.handler, "handlerMetrics", DefaultMessageHandlerMetrics.class); - handlerMetrics = Mockito.spy(handlerMetrics); + DefaultMessageChannelMetrics channelMetrics = + TestUtils.getPropertyValue(this.next, "channelMetrics", DefaultMessageChannelMetrics.class); + channelMetrics = Mockito.spy(channelMetrics); Mockito.doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Object result = invocation.callRealMethod(); - afterHandleLatch.countDown(); + afterSendLatch.countDown(); return result; } - }).when(handlerMetrics).afterHandle(Mockito.any(MetricsContext.class), Mockito.eq(Boolean.TRUE)); + }).when(channelMetrics).afterSend(Mockito.any(MetricsContext.class), Mockito.eq(Boolean.TRUE)); - new DirectFieldAccessor(this.handler).setPropertyValue("handlerMetrics", handlerMetrics); + new DirectFieldAccessor(this.next).setPropertyValue("channelMetrics", channelMetrics); MessagingTemplate messagingTemplate = new MessagingTemplate(this.input); messagingTemplate.setReceiveTimeout(10000); Integer active = messagingTemplate.convertSendAndReceive("foo", Integer.class); assertEquals(1, active.intValue()); - assertTrue(afterHandleLatch.await(10, TimeUnit.SECONDS)); + assertTrue(afterSendLatch.await(10, TimeUnit.SECONDS)); assertEquals(0, this.handler.getActiveCount()); assertEquals(1, this.handler.getHandleCount()); assertThat(this.handler.getDuration().getMax(), greaterThan(99.0)); diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java index 95158b3b0d..d72fb18367 100644 --- a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java +++ b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java @@ -20,7 +20,10 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Date; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -39,6 +42,7 @@ import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandler; import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -72,6 +76,8 @@ public abstract class AbstractStompSessionManager implements StompSessionManager protected final StompClientSupport stompClient; + private final AtomicInteger epoch = new AtomicInteger(); + private boolean autoStartup = false; private boolean running = false; @@ -164,18 +170,43 @@ public abstract class AbstractStompSessionManager implements StompSessionManager return this.phase; } - private void connect() { + private synchronized void connect() { + if (this.connecting || this.connected) { + if (logger.isDebugEnabled()) { + logger.debug("Aborting connect; another thread is connecting."); + } + return; + } + final int epoch = this.epoch.get(); this.connecting = true; - this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler); + if (logger.isDebugEnabled()) { + logger.debug("Connecting " + this); + } + try { + this.stompSessionListenableFuture = doConnect(this.compositeStompSessionHandler); + } + catch (Exception e) { + logger.error("doConnect() error for " + this, e); + } + final CountDownLatch latch = new CountDownLatch(1); this.stompSessionListenableFuture.addCallback(new ListenableFutureCallback() { @Override public void onFailure(Throwable e) { - scheduleReconnect(e); + if (logger.isDebugEnabled()) { + logger.debug("onFailure", e); + } + latch.countDown(); + if (epoch == AbstractStompSessionManager.this.epoch.get()) { + scheduleReconnect(e); + } } @Override public void onSuccess(StompSession stompSession) { + if (logger.isDebugEnabled()) { + logger.debug("onSuccess"); + } AbstractStompSessionManager.this.connected = true; AbstractStompSessionManager.this.connecting = false; stompSession.setAutoReceipt(isAutoReceiptEnabled()); @@ -184,14 +215,28 @@ public abstract class AbstractStompSessionManager implements StompSessionManager new StompSessionConnectedEvent(this)); } AbstractStompSessionManager.this.reconnectFuture = null; + latch.countDown(); } }); + try { + if (!latch.await(10, TimeUnit.SECONDS)) { + logger.error("No response to connection attempt"); + if (epoch == this.epoch.get()) { + scheduleReconnect(null); + } + } + } + catch (InterruptedException e1) { + logger.error("Interrupted while waiting for connection attempt"); + Thread.currentThread().interrupt(); + } } private void scheduleReconnect(Throwable e) { + this.epoch.incrementAndGet(); this.connecting = this.connected = false; - logger.error("STOMP connect error.", e); + logger.error("STOMP connect error for " + this, e); if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent( new StompConnectionFailedEvent(this, e)); @@ -297,8 +342,9 @@ public abstract class AbstractStompSessionManager implements StompSessionManager @Override public String toString() { - return "StompSessionManager{" + - "connected=" + connected + + return ObjectUtils.identityToString(this) + + " {connecting=" + connecting + + ", connected=" + connected + ", name='" + name + '\'' + '}'; } diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java index 00950da56e..3ec3f4218b 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java @@ -75,7 +75,8 @@ public class StompServerIntegrationTests extends LogAdjustingTestSupport { public StompServerIntegrationTests() { - super("org.springframework", "org.springframework.integration.stomp"); + super("org.springframework", "org.springframework.integration.stomp", + "org.apache.activemq.broker", "reactor.io", "io.netty"); } @BeforeClass