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

Fix TCP Cache Race Condition

Browse Source 
On a gateway remote timeout, there's a race between the reader and writer
thread to close the connection.

Change the `released` boolean to an `AtomicBoolean` to avoid the unexpected
exception when the other thread attempts to return a non-existent connection
to the pool.
This commit is contained in:
Gary Russell
2016-02-16 15:48:40 -05:00
parent 96565cc716
commit d48597f41b
2 changed files with 6 additions and 6 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/connection/CachingClientConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2015 the original author or authors. * Copyright 2002-2016 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer; import org.springframework.core.serializer.Serializer;
@@ -137,7 +138,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
private class CachedConnection extends TcpConnectionInterceptorSupport { private class CachedConnection extends TcpConnectionInterceptorSupport {
private volatile boolean released; private final AtomicBoolean released = new AtomicBoolean();
private CachedConnection(TcpConnectionSupport connection, TcpListener tcpListener) { private CachedConnection(TcpConnectionSupport connection, TcpListener tcpListener) {
super.setTheConnection(connection); super.setTheConnection(connection);
@@ -146,7 +147,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
@Override @Override
public void close() { public void close() {
if (this.released) { if (!this.released.compareAndSet(false, true)) {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Connection " + getConnectionId() + " has already been released"); logger.debug("Connection " + getConnectionId() + " has already been released");
} }
@@ -163,7 +164,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
super.close(); super.close();
} }
pool.releaseItem(getTheConnection()); pool.releaseItem(getTheConnection());
this.released = true;
} }
} }

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2015 the original author or authors. * Copyright 2002-2016 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -442,7 +442,7 @@ public class TcpOutboundGatewayTests extends LogAdjustingTestSupport {
} }
else { else {
assertNotNull(e.getCause()); assertNotNull(e.getCause());
assertTrue(e.getCause() instanceof MessageTimeoutException); assertThat(e.getCause(), instanceOf(MessageTimeoutException.class));
} }
timeouts++; timeouts++;
continue; continue;