From bf4005a92c42d4a69810bbdc363e7ee055034f01 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/TcpConnectionSupport.java | 3 +++ .../TcpNioServerConnectionFactory.java | 26 ++++++++++++------- .../integration/ip/util/TestingUtilities.java | 22 +++++++++++++++- .../FailoverClientConnectionFactoryTests.java | 3 ++- 4 files changed, 42 insertions(+), 12 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 53fc5ce86d..994e0165d7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -120,6 +120,9 @@ public abstract class TcpConnectionSupport implements TcpConnection { this.connectionFactoryName = connectionFactoryName; } this.publishConnectionOpenEvent(); + if (logger.isDebugEnabled()) { + logger.debug("New connection " + this); + } } 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 7485c5d614..aa882c5e46 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..7d5960947a 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,6 +15,7 @@ */ package org.springframework.integration.ip.util; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; /** @@ -88,4 +89,23 @@ 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; + while (timer < 10000) { + if (factory.getOpenConnectionIds().size() == n) { + return; + } + Thread.sleep(100); + timer += 100; + } + throw new IllegalStateException("Connections=" + factory.getOpenConnectionIds().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 ab473e2786..9f7721b7cf 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 @@ -335,7 +335,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 +343,7 @@ public class FailoverClientConnectionFactoryTests { replyMessage = replyChannel.receive(10000); assertNotNull(replyMessage); gateway2.stop(); + outGateway.stop(); } private Socket getSocket(AbstractClientConnectionFactory client) throws Exception {