DATAREDIS-637 - Shut down connection-exclusive ClusterCommandExecutor.

We now shut down ClusterCommandExecutor via LettuceClusterConnection and JedisClusterConnection if the ClusterCommandExecutor was created through the connection instance. In such case ClusterCommandExecutor is managed through Lettuce/JedisClusterConnection and disposed on connection close.

Previously ClusterCommandExecutor created through a ClusterConnection was not disposed.

Original pull request: #247.

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -31,6 +31,8 @@ import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -101,18 +103,21 @@ public class JedisClusterConnection implements RedisClusterConnection {
 	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy(
 			JedisConverters.exceptionConverter());
 
+	private final Log log = LogFactory.getLog(getClass());
+
 	private final JedisCluster cluster;
 
 	private boolean closed;
 
 	private final JedisClusterTopologyProvider topologyProvider;
 	private ClusterCommandExecutor clusterCommandExecutor;
+	private final boolean disposeClusterCommandExecutorOnClose;
 
 	private volatile JedisSubscription subscription;
 
 	/**
 	 * Create new {@link JedisClusterConnection} utilizing native connections via {@link JedisCluster}.
-	 * 
+	 *
 	 * @param cluster must not be {@literal null}.
 	 */
 	public JedisClusterConnection(JedisCluster cluster) {
@@ -125,6 +130,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
 		topologyProvider = new JedisClusterTopologyProvider(cluster);
 		clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new JedisClusterNodeResourceProvider(cluster),
 				EXCEPTION_TRANSLATION);
+		disposeClusterCommandExecutorOnClose = true;
 
 		try {
 			DirectFieldAccessor dfa = new DirectFieldAccessor(cluster);
@@ -151,6 +157,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
 		this.cluster = cluster;
 		this.topologyProvider = new JedisClusterTopologyProvider(cluster);
 		this.clusterCommandExecutor = executor;
+		this.disposeClusterCommandExecutorOnClose = false;
 	}
 
 	/*
@@ -3998,6 +4005,15 @@ public class JedisClusterConnection implements RedisClusterConnection {
 	 */
 	@Override
 	public void close() throws DataAccessException {
+
+		if (!closed && disposeClusterCommandExecutorOnClose) {
+			try {
+				clusterCommandExecutor.destroy();
+			} catch (Exception ex) {
+				log.warn("Cannot properly close cluster command executor", ex);
+			}
+		}
+
 		closed = true;
 	}
 

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java

@@ -29,8 +29,11 @@ import java.util.Properties;
 import java.util.Random;
 import java.util.Set;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
@@ -79,13 +82,16 @@ public class LettuceClusterConnection extends LettuceConnection
 			new LettuceExceptionConverter());
 	static final RedisCodec<byte[], byte[]> CODEC = ByteArrayCodec.INSTANCE;
 
+	private final Log log = LogFactory.getLog(getClass());
+
 	private final RedisClusterClient clusterClient;
 	private ClusterCommandExecutor clusterCommandExecutor;
 	private ClusterTopologyProvider topologyProvider;
+	private final boolean disposeClusterCommandExecutorOnClose;
 
 	/**
 	 * Creates new {@link LettuceClusterConnection} using {@link RedisClusterClient}.
-	 * 
+	 *
 	 * @param clusterClient must not be {@literal null}.
 	 */
 	public LettuceClusterConnection(RedisClusterClient clusterClient) {
@@ -98,6 +104,7 @@ public class LettuceClusterConnection extends LettuceConnection
 		topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
 		clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
 				new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter);
+		disposeClusterCommandExecutorOnClose = true;
 	}
 
 	/**
@@ -117,6 +124,7 @@ public class LettuceClusterConnection extends LettuceConnection
 		this.clusterClient = clusterClient;
 		topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
 		clusterCommandExecutor = executor;
+		disposeClusterCommandExecutorOnClose = false;
 	}
 
 	/*
@@ -1513,6 +1521,24 @@ public class LettuceClusterConnection extends LettuceConnection
 		return result;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisConnection#close()
+	 */
+	@Override
+	public void close() throws DataAccessException {
+
+		if (!isClosed() && disposeClusterCommandExecutorOnClose) {
+			try {
+				clusterCommandExecutor.destroy();
+			} catch (Exception ex) {
+				log.warn("Cannot properly close cluster command executor", ex);
+			}
+		}
+
+		super.close();
+	}
+
 	/**
 	 * Lettuce specific implementation of {@link ClusterCommandCallback}.
 	 *

src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java

@@ -49,6 +49,7 @@ import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
 import org.springframework.data.redis.core.RedisKeyValueAdapter.EnableKeyspaceEvents;
 import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
 import org.springframework.data.redis.core.convert.MappingConfiguration;
@@ -82,7 +83,10 @@ public class RedisKeyValueAdapterTests {
 
 	@Parameters
 	public static List<RedisConnectionFactory> params() {
-		return Arrays.<RedisConnectionFactory> asList(new JedisConnectionFactory(), new LettuceConnectionFactory());
+
+		LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory();
+		lettuceConnectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources());
+		return Arrays.<RedisConnectionFactory> asList(new JedisConnectionFactory(), lettuceConnectionFactory);
 	}
 
 	@AfterClass
@@ -113,6 +117,14 @@ public class RedisKeyValueAdapterTests {
 				return null;
 			}
 		});
+
+		RedisConnection connection = template.getConnectionFactory().getConnection();
+
+		try {
+			connection.setConfig("notify-keyspace-events", "KEA");
+		} finally {
+			connection.close();
+		}
 	}
 
 	@After