From d48597f41b194135733a581373af1d872fbe8c7e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 16 Feb 2016 15:48:40 -0500 Subject: [PATCH] Fix TCP Cache Race Condition 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. --- .../ip/tcp/connection/CachingClientConnectionFactory.java | 8 ++++---- .../integration/ip/tcp/TcpOutboundGatewayTests.java | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index e26731bfbf..e42903ec00 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -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"); * 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.Map; import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -137,7 +138,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact private class CachedConnection extends TcpConnectionInterceptorSupport { - private volatile boolean released; + private final AtomicBoolean released = new AtomicBoolean(); private CachedConnection(TcpConnectionSupport connection, TcpListener tcpListener) { super.setTheConnection(connection); @@ -146,7 +147,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public void close() { - if (this.released) { + if (!this.released.compareAndSet(false, true)) { if (logger.isDebugEnabled()) { logger.debug("Connection " + getConnectionId() + " has already been released"); } @@ -163,7 +164,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact super.close(); } pool.releaseItem(getTheConnection()); - this.released = true; } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 3d2a84fa39..3a91365197 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -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"); * you may not use this file except in compliance with the License. @@ -442,7 +442,7 @@ public class TcpOutboundGatewayTests extends LogAdjustingTestSupport { } else { assertNotNull(e.getCause()); - assertTrue(e.getCause() instanceof MessageTimeoutException); + assertThat(e.getCause(), instanceOf(MessageTimeoutException.class)); } timeouts++; continue;