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

@@ -30,6 +30,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;
@@ -91,18 +93,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) {
@@ -115,6 +120,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);
@@ -141,6 +147,7 @@ public class JedisClusterConnection implements RedisClusterConnection {
 		this.cluster = cluster;
 		this.topologyProvider = new JedisClusterTopologyProvider(cluster);
 		this.clusterCommandExecutor = executor;
+		this.disposeClusterCommandExecutorOnClose = false;
 	}
 
 	/*
@@ -3702,6 +3709,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.dao.DataAccessResourceFailureException;
+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 = new BytesRedisCodec();
 
+	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;
 	}
 
 	/*
@@ -1539,6 +1547,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}.
 	 *