From 5371ead6771aede622065790c8e7d16f7c2a7c12 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 26 Jun 2012 08:00:33 -0400 Subject: [PATCH] INT-2620 Fix Race Condition in Failover Tests Sporadic test failures (fairly consistently on Mac). Problem was some tests sent a message to a client socket before the close notification had been received. Wait for client connection to close rather than waiting for the server to stop. Also, with NIO, close the channel on accept exception. Also improve debugging by including a log of new connections including the id. --- .../tcp/connection/AbstractTcpConnection.java | 6 +++- .../TcpNioServerConnectionFactory.java | 26 ++++++++------ .../integration/ip/util/TestingUtilities.java | 35 ++++++++++++++++++- .../FailoverClientConnectionFactoryTests.java | 6 ++-- 4 files changed, 59 insertions(+), 14 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 44e5e58375..5db643e5a7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-2013 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. @@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; @@ -87,6 +88,9 @@ public abstract class AbstractTcpConnection implements TcpConnection { try { this.soLinger = socket.getSoLinger(); } catch (SocketException e) { } + if (logger.isDebugEnabled()) { + logger.debug("New connection " + this.getConnectionId()); + } } public void afterSend(Message message) throws Exception { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 78fa196f07..d7b85a8c82 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -156,17 +156,23 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto channel.close(); } else { - channel.configureBlocking(false); - Socket socket = channel.socket(); - setSocketAttributes(socket); - TcpNioConnection connection = createTcpNioConnection(channel); - if (connection == null) { - return; + try { + channel.configureBlocking(false); + Socket socket = channel.socket(); + setSocketAttributes(socket); + TcpNioConnection connection = createTcpNioConnection(channel); + if (connection == null) { + return; + } + connection.setTaskExecutor(this.getTaskExecutor()); + connection.setLastRead(now); + this.channelMap.put(channel, connection); + channel.register(selector, SelectionKey.OP_READ, connection); + } + catch (Exception e) { + logger.error("Exception accepting new connection", e); + channel.close(); } - connection.setTaskExecutor(this.getTaskExecutor()); - connection.setLastRead(now); - this.channelMap.put(channel, connection); - channel.register(selector, SelectionKey.OP_READ, connection); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java index 671b4ab608..d01a7085ad 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/util/TestingUtilities.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -15,7 +15,12 @@ */ package org.springframework.integration.ip.util; +import java.util.Collection; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnection; /** * Convenience class providing methods for testing IP components. @@ -88,4 +93,32 @@ public class TestingUtilities { } } + /** + * Wait for up to 10 seconds for the connection factory to have the specified number + * of connections. + * @param factory The factory. + * @param n The required number of connections. + * @throws Exception IllegalStateException if the count does not match. + */ + public static void waitUntilFactoryHasThisNumberOfConnections(AbstractConnectionFactory factory, int n) + throws Exception { + int timer = 0; + DirectFieldAccessor accessor = new DirectFieldAccessor(factory); + @SuppressWarnings("unchecked") + Collection connections = (Collection) accessor.getPropertyValue("connections"); + while (timer < 10000) { + int open = 0; + for (TcpConnection connection : connections) { + if (connection.isOpen()) { + open++; + } + } + if (open == n) { + return; + } + Thread.sleep(100); + timer += 100; + } + throw new IllegalStateException("Connections=" + connections.size() + "wanted=" + n); + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java index fd607c6470..143515898b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -39,6 +39,7 @@ import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; @@ -335,7 +336,7 @@ public class FailoverClientConnectionFactoryTests { Message replyMessage = replyChannel.receive(10000); assertNotNull(replyMessage); server1.stop(); - TestingUtilities.waitStopListening(server1, null); + TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(client1, 0); outGateway.handleMessage(message); socket = getSocket(client2); port2 = socket.getLocalPort(); @@ -343,6 +344,7 @@ public class FailoverClientConnectionFactoryTests { replyMessage = replyChannel.receive(10000); assertNotNull(replyMessage); gateway2.stop(); + outGateway.stop(); } private Socket getSocket(AbstractClientConnectionFactory client) throws Exception {