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

INT-3123 TCP Fix Error Propagation with Failover

Browse Source 
The second chance NIO logic that detects a close being received
in a race condition with a reply needs to be enabled always because
a Failover or Caching connection factory might be using NIO
underneath.

Also, add an epoch to the FailoverTcpConnection's connectionId which
is incremented on each use to prevent a close after a previous
use being received by the current user.

Finally, add code in the FailoverTcpConnection to not propagate
messages from an old (defunct) delegate connection - if the
delegate has changed, the connection must be being used by
another client.
This commit is contained in:
Gary Russell
2013-08-29 14:29:30 -04:00
parent 7dabe1334c
commit 9285867465
5 changed files with 55 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -33,7 +33,6 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.util.Assert;
@@ -138,10 +137,10 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
return replyMessage;
}
catch (Exception e) {
logger.error("Tcp Gateway exception", e);
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
logger.error("Tcp Gateway exception", e);
throw new MessagingException("Failed to send or receive", e);
}
finally {
@@ -163,6 +162,9 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
logger.error("Cannot correlate response - no connection id");
return false;
}
if (logger.isTraceEnabled()) {
logger.trace("onMessage: " + connectionId + "(" + message + ")");
}
AsyncReply reply = pendingReplies.get(connectionId);
if (reply == null) {
if (message instanceof ErrorMessage) {
@@ -282,7 +284,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
boolean waitForMessageAfterError = connectionFactory instanceof TcpNioClientConnectionFactory;
boolean waitForMessageAfterError = true;
while (reply instanceof ErrorMessage) {
if (waitForMessageAfterError) {
/*

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -22,6 +22,7 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.SimplePool;
@@ -317,7 +318,10 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
super.registerListener(listener);
targetConnectionFactory.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
throw new UnsupportedOperationException("This should never be called");
if (!(message instanceof ErrorMessage)) {
throw new UnsupportedOperationException("This should never be called");
}
return false;
}
});
}

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

@@ -19,11 +19,13 @@ import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
@@ -81,7 +83,10 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
for (AbstractClientConnectionFactory factory : this.factories) {
factory.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
throw new UnsupportedOperationException("This should never be called");
if (!(message instanceof ErrorMessage)) {
throw new UnsupportedOperationException("This should never be called");
}
return false;
}
});
}
@@ -98,10 +103,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
protected TcpConnectionSupport obtainConnection() throws Exception {
TcpConnectionSupport connection = this.getTheConnection();
if (connection != null && connection.isOpen()) {
((FailoverTcpConnection) connection).incrementEpoch();
return connection;
}
FailoverTcpConnection failoverTcpConnection = new FailoverTcpConnection(this.factories);
failoverTcpConnection.registerListener(this.getListener());
failoverTcpConnection.incrementEpoch();
return failoverTcpConnection;
}
@@ -162,6 +169,8 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
private volatile boolean open = true;
private final AtomicLong epoch = new AtomicLong();
public FailoverTcpConnection(List<AbstractClientConnectionFactory> factories) throws Exception {
this.factories = factories;
this.factoryIterator = factories.iterator();
@@ -169,6 +178,10 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.connectionId = UUID.randomUUID().toString();
}
void incrementEpoch() {
this.epoch.incrementAndGet();
}
/**
* Finds a connection from the underlying list of factories. If necessary,
* each factory is tried; including the current one if we wrap around.
@@ -288,7 +301,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
@Override
public String getConnectionId() {
return this.connectionId;
return this.connectionId + ":" + epoch;
}
@Override
@@ -331,11 +344,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.delegate.setSerializer(serializer);
}
@Override
public long incrementAndGetConnectionSequence() {
return this.delegate.incrementAndGetConnectionSequence();
}
/**
* We have to intercept the message to replace the connectionId header with
* ours so the listener can correlate a response with a request. We supply
@@ -343,13 +351,21 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* purposes.
*/
public boolean onMessage(Message<?> message) {
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
if (this.delegate.getConnectionId().equals(message.getHeaders().get(IpHeaders.CONNECTION_ID))) {
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
return this.getListener().onMessage(messageBuilder.build());
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Message from defunct connection ignored " + message);
}
return false;
}
return this.getListener().onMessage(messageBuilder.build());
}
}

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

@@ -382,8 +382,10 @@ public class TcpNioConnection extends TcpConnectionSupport {
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Running an assembler");
}
this.taskExecutor.execute(this);
logger.debug("Running an assembler");
} else {
this.executionControl.decrementAndGet();
}

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

@@ -39,6 +39,9 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
@@ -296,6 +299,16 @@ public class FailoverClientConnectionFactoryTests {
client2.setTaskExecutor(exec);
server1.setTaskExecutor(exec);
server2.setTaskExecutor(exec);
ApplicationEventPublisher pub = new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
};
client1.setApplicationEventPublisher(pub);
client2.setApplicationEventPublisher(pub);
server1.setApplicationEventPublisher(pub);
server2.setApplicationEventPublisher(pub);
TcpInboundGateway gateway1 = new TcpInboundGateway();
gateway1.setConnectionFactory(server1);
SubscribableChannel channel = new DirectChannel();