Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3795: TCP: Combo Failover and Cached CFs

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3795

Previously, a `CachingClientConnectionFactory` could wrap `FailoverClientConnectionFactory` instances.

It was not possible to wrap `CachingClientConnectionFactory` instances in a `FailoverClientConnectionFactory`.

The failover logic only failed over when an `IOException` occurred, whereas the CCCF wrapped the IOException
in a `MessagingException`.

- Catch all exceptions in the failover logic
- Add real TCP test cases for both scenarios
 - test failover when a connection is closed
 - test failover when the first factory cannot connect at all

INT-3795: Polishing - Fix Test Race

Revert `.travis.yml` changes
This commit is contained in:
Gary Russell
2015-08-05 15:30:12 -04:00
committed by Artem Bilan
parent 2376a558a4
commit 0ce5f12342
5 changed files with 369 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -15,7 +15,6 @@
*/
package org.springframework.integration.ip.tcp.connection;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
@@ -177,7 +176,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* This allows for the condition where the current connection is closed,
* the current factory can serve up a new connection, but all other
* factories are down.
* @throws Exception
* @throws Exception if an exception occurs
*/
private synchronized void findAConnection() throws Exception {
boolean success = false;
@@ -191,11 +190,19 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
try {
nextFactory = this.factoryIterator.next();
this.delegate = nextFactory.getConnection();
if (logger.isDebugEnabled()) {
logger.debug("Got " + this.delegate.getConnectionId() + " from " + nextFactory);
}
this.delegate.registerListener(this);
this.currentFactory = nextFactory;
success = this.delegate.isOpen();
}
catch (IOException e) {
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug(nextFactory + " failed with "
+ e.toString()
+ ", trying another");
}
if (!this.factoryIterator.hasNext()) {
if (retried && lastFactoryToTry == null || lastFactoryToTry == nextFactory) {
/*
@@ -240,7 +247,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.delegate.send(message);
success = true;
}
catch (IOException e) {
catch (Exception e) {
if (retried && lastFactoryTried == lastFactoryToTry) {
logger.error("All connection factories exhausted", e);
this.open = false;

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -111,7 +111,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
throw e;
}
if (logger.isDebugEnabled()) {
logger.debug("Message sent " + message);
logger.debug(getConnectionId() + " Message sent " + message);
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -156,7 +156,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
throw e;
}
if (logger.isDebugEnabled()) {
logger.debug("Message sent " + message);
logger.debug(getConnectionId() + " Message sent " + message);
}
}
}

165
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -75,6 +75,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.SimplePool;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
@@ -567,6 +568,170 @@ public class CachingClientConnectionFactoryTests {
Mockito.verify(mockConn2).send(message);
}
@Test
public void testCachedFailoverRealClose() throws Exception {
int port1 = SocketUtils.findAvailableTcpPort();
TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
server1.setBeanName("server1");
final CountDownLatch latch1 = new CountDownLatch(3);
server1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch1.countDown();
return false;
}
});
server1.start();
TestingUtilities.waitListening(server1, 10000L);
int port2 = SocketUtils.findAvailableTcpPort();
TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
server1.setBeanName("server2");
final CountDownLatch latch2 = new CountDownLatch(2);
server2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch2.countDown();
return false;
}
});
server2.start();
TestingUtilities.waitListening(server2, 10000L);
// Failover
AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("localhost", port1);
factory1.setBeanName("client1");
factory1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2);
factory2.setBeanName("client2");
factory2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(factory1);
factories.add(factory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
// Cache
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2);
cachingFactory.start();
TcpConnection conn1 = cachingFactory.getConnection();
GenericMessage<String> message = new GenericMessage<String>("foo");
conn1.send(message);
conn1.close();
TcpConnection conn2 = cachingFactory.getConnection();
assertSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(),
((TcpConnectionInterceptorSupport) conn2).getTheConnection());
conn2.send(message);
conn1 = cachingFactory.getConnection();
assertNotSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(),
((TcpConnectionInterceptorSupport) conn2).getTheConnection());
conn1.send(message);
conn1.close();
conn2.close();
assertTrue(latch1.await(10, TimeUnit.SECONDS));
server1.stop();
TestingUtilities.waitStopListening(server1, 10000L);
TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(factory1, 0);
conn1 = cachingFactory.getConnection();
conn2 = cachingFactory.getConnection();
conn1.send(message);
conn2.send(message);
conn1.close();
conn2.close();
assertTrue(latch2.await(10, TimeUnit.SECONDS));
SimplePool<?> pool = TestUtils.getPropertyValue(cachingFactory, "pool", SimplePool.class);
assertEquals(2, pool.getIdleCount());
server2.stop();
}
@Test
public void testCachedFailoverRealBadHost() throws Exception {
int port1 = SocketUtils.findAvailableTcpPort();
TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
server1.setBeanName("server1");
final CountDownLatch latch1 = new CountDownLatch(3);
server1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch1.countDown();
return false;
}
});
server1.start();
TestingUtilities.waitListening(server1, 10000L);
int port2 = SocketUtils.findAvailableTcpPort();
TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
server1.setBeanName("server2");
final CountDownLatch latch2 = new CountDownLatch(2);
server2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch2.countDown();
return false;
}
});
server2.start();
TestingUtilities.waitListening(server2, 10000L);
// Failover
AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("junkjunk", port1);
factory1.setBeanName("client1");
factory1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2);
factory2.setBeanName("client2");
factory2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(factory1);
factories.add(factory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
// Cache
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2);
cachingFactory.start();
TcpConnection conn1 = cachingFactory.getConnection();
GenericMessage<String> message = new GenericMessage<String>("foo");
conn1.send(message);
conn1.close();
TcpConnection conn2 = cachingFactory.getConnection();
assertSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(),
((TcpConnectionInterceptorSupport) conn2).getTheConnection());
conn2.send(message);
conn1 = cachingFactory.getConnection();
assertNotSame(((TcpConnectionInterceptorSupport) conn1).getTheConnection(),
((TcpConnectionInterceptorSupport) conn2).getTheConnection());
conn1.send(message);
conn1.close();
conn2.close();
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertEquals(3, latch1.getCount());
server1.stop();
server2.stop();
}
@Test //INT-3650
public void testRealConnection() throws Exception {
int port = SocketUtils.findAvailableTcpPort();

192
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -16,7 +16,10 @@
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
@@ -30,8 +33,10 @@ import java.net.Socket;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -55,6 +60,7 @@ import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.test.rule.Log4jLevelAdjuster;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.SimplePool;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@@ -71,7 +77,7 @@ public class FailoverClientConnectionFactoryTests {
@Rule
public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE,
"org.springframework.integration.ip.tcp");
"org.springframework.integration.ip.tcp", "org.springframework.integration.util.SimplePool");
@Test
public void testFailoverGood() throws Exception {
@@ -303,6 +309,190 @@ public class FailoverClientConnectionFactoryTests {
testRealGuts(client1, client2, server1, server2);
}
@Test
public void testFailoverCachedRealClose() throws Exception {
int port1 = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
server1.setBeanName("server1");
final CountDownLatch latch1 = new CountDownLatch(3);
server1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch1.countDown();
return false;
}
});
server1.start();
TestingUtilities.waitListening(server1, 10000L);
int port2 = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
server2.setBeanName("server2");
final CountDownLatch latch2 = new CountDownLatch(2);
server2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch2.countDown();
return false;
}
});
server2.start();
TestingUtilities.waitListening(server2, 10000L);
AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("localhost", port1);
factory1.setBeanName("client1");
factory1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2);
factory2.setBeanName("client2");
factory2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
// Cache
CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 2);
cachingFactory1.setBeanName("cache1");
CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 2);
cachingFactory2.setBeanName("cache2");
// Failover
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(cachingFactory1);
factories.add(cachingFactory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
TcpConnection conn1 = failoverFactory.getConnection();
conn1.send(new GenericMessage<String>("foo1"));
conn1.close();
TcpConnection conn2 = failoverFactory.getConnection();
assertSame(
(TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection(),
(TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection());
conn2.send(new GenericMessage<String>("foo2"));
conn1 = failoverFactory.getConnection();
assertNotSame(
(TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection(),
(TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection());
conn1.send(new GenericMessage<String>("foo3"));
conn1.close();
conn2.close();
assertTrue(latch1.await(10, TimeUnit.SECONDS));
server1.stop();
TestingUtilities.waitStopListening(server1, 10000L);
TestingUtilities.waitUntilFactoryHasThisNumberOfConnections(factory1, 0);
conn1 = failoverFactory.getConnection();
conn2 = failoverFactory.getConnection();
conn1.send(new GenericMessage<String>("foo4"));
conn2.send(new GenericMessage<String>("foo5"));
conn1.close();
conn2.close();
assertTrue(latch2.await(10, TimeUnit.SECONDS));
SimplePool<?> pool = TestUtils.getPropertyValue(cachingFactory2, "pool", SimplePool.class);
assertEquals(2, pool.getIdleCount());
server2.stop();
}
@Test
public void testFailoverCachedRealBadHost() throws Exception {
int port1 = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server1 = new TcpNetServerConnectionFactory(port1);
server1.setBeanName("server1");
final CountDownLatch latch1 = new CountDownLatch(3);
server1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch1.countDown();
return false;
}
});
server1.start();
TestingUtilities.waitListening(server1, 10000L);
int port2 = SocketUtils.findAvailableServerSocket();
TcpNetServerConnectionFactory server2 = new TcpNetServerConnectionFactory(port2);
server2.setBeanName("server2");
final CountDownLatch latch2 = new CountDownLatch(2);
server2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
latch2.countDown();
return false;
}
});
server2.start();
TestingUtilities.waitListening(server2, 10000L);
AbstractClientConnectionFactory factory1 = new TcpNetClientConnectionFactory("junkjunk", port1);
factory1.setBeanName("client1");
factory1.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port2);
factory2.setBeanName("client2");
factory2.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
return false;
}
});
// Cache
CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 2);
cachingFactory1.setBeanName("cache1");
CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 2);
cachingFactory2.setBeanName("cache2");
// Failover
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(cachingFactory1);
factories.add(cachingFactory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
TcpConnection conn1 = failoverFactory.getConnection();
GenericMessage<String> message = new GenericMessage<String>("foo");
conn1.send(message);
conn1.close();
TcpConnection conn2 = failoverFactory.getConnection();
assertSame(
(TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection(),
(TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection());
conn2.send(message);
conn1 = failoverFactory.getConnection();
assertNotSame(
(TestUtils.getPropertyValue(conn1, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection(),
(TestUtils.getPropertyValue(conn2, "delegate", TcpConnectionInterceptorSupport.class))
.getTheConnection());
conn1.send(message);
conn1.close();
conn2.close();
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertEquals(3, latch1.getCount());
server1.stop();
server2.stop();
}
private void testRealGuts(AbstractClientConnectionFactory client1, AbstractClientConnectionFactory client2,
AbstractServerConnectionFactory server1, AbstractServerConnectionFactory server2) throws Exception {
int port1 = 0;