INT-3722: Additional Fix: TCP OG with Caching CF
JIRA: https://jira.spring.io/browse/INT-3722 While the initial fix solved the reported problem, it caused a memory leak when the gateway completed its work before the reader thread "closed" the connection, releasing it to the pool. This left the connection in the deferred close state. Handle the condition where the gateway signals it has completed its work before the connection is released to the pool.
This commit is contained in:
@@ -16,7 +16,9 @@
|
||||
|
||||
package org.springframework.integration.ip.tcp.connection;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
@@ -48,6 +50,8 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
|
||||
private final Map<String, CachedConnection> deferredClosures =
|
||||
new ConcurrentHashMap<String, CachedConnection>();
|
||||
|
||||
private final Set<String> okToRelease = new HashSet<String>();
|
||||
|
||||
private volatile boolean deferClose;
|
||||
|
||||
/**
|
||||
@@ -149,9 +153,14 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
|
||||
|
||||
@Override
|
||||
public void closeDeferred(String connectionId) {
|
||||
CachedConnection deferred = this.deferredClosures.remove(connectionId);
|
||||
if (deferred != null) {
|
||||
deferred.doClose();
|
||||
synchronized(this.okToRelease) {
|
||||
CachedConnection deferred = this.deferredClosures.remove(connectionId);
|
||||
if (deferred != null) {
|
||||
deferred.doClose();
|
||||
}
|
||||
else {
|
||||
this.okToRelease.add(connectionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -165,12 +174,14 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void close() {
|
||||
if (deferClose && !this.released) {
|
||||
deferredClosures.put(getConnectionId(), this);
|
||||
}
|
||||
else {
|
||||
doClose();
|
||||
public void close() {
|
||||
synchronized(okToRelease) {
|
||||
if (deferClose && !this.released && !okToRelease.remove(getConnectionId())) {
|
||||
deferredClosures.put(getConnectionId(), this);
|
||||
}
|
||||
else {
|
||||
doClose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -457,6 +457,7 @@ public class CachingClientConnectionFactoryTests {
|
||||
}
|
||||
|
||||
@Test
|
||||
// @Repeat(1000) // INT-3722
|
||||
public void gatewayIntegrationTest() throws Exception {
|
||||
final List<String> connectionIds = new ArrayList<String>();
|
||||
final AtomicBoolean okToRun = new AtomicBoolean(true);
|
||||
|
||||
Reference in New Issue
Block a user