DATAREDIS-731 - Share native Lettuce Cluster connection across LettuceClusterConnection.

We now share Lettuce Cluster connections if configured on LettuceConnectionFactory (enabled by default) across multiple LettuceClusterConnections. Connection sharing prevents creation and disposal of cluster connections per template command execution which improves overall performance.

Original Pull Request: #297

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

@@ -152,6 +152,28 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 		this.disposeClusterCommandExecutorOnClose = false;
 	}
 
+	/**
+	 * Creates new {@link LettuceClusterConnection} given a shared {@link StatefulRedisClusterConnection}
+	 * and{@link LettuceConnectionProvider} running commands across the cluster via given {@link ClusterCommandExecutor}.
+	 *
+	 * @param sharedConnection must not be {@literal null}.
+	 * @param connectionProvider must not be {@literal null}.
+	 * @param executor must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public LettuceClusterConnection(StatefulRedisClusterConnection<byte[], byte[]> sharedConnection,
+			LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor, Duration timeout) {
+
+		super(sharedConnection, connectionProvider, timeout.toMillis(), 0);
+
+		Assert.notNull(executor, "ClusterCommandExecutor must not be null.");
+
+		this.topologyProvider = new LettuceClusterTopologyProvider(getClient());
+		this.clusterCommandExecutor = executor;
+		this.disposeClusterCommandExecutorOnClose = false;
+	}
+
 	/**
 	 * @return access to {@link RedisClusterClient} for non-connection access.
 	 */
@@ -319,7 +341,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 
 		clusterCommandExecutor.executeCommandOnSingleNode(
 				(LettuceClusterCommandCallback<String>) client -> client.clusterAddSlots(slots), node);
-
 	}
 
 	/*

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

@@ -243,6 +243,19 @@ public class LettuceConnection extends AbstractRedisConnection {
 	 */
 	public LettuceConnection(StatefulRedisConnection<byte[], byte[]> sharedConnection,
 			LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) {
+		this((StatefulConnection<byte[], byte[]>) sharedConnection, connectionProvider, timeout, defaultDbIndex);
+	}
+
+	/**
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations.
+	 * @param connectionProvider connection provider to obtain and release native connections.
+	 * @param timeout The connection timeout (in milliseconds)
+	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
+	 * @since 2.1
+	 */
+	LettuceConnection(StatefulConnection<byte[], byte[]> sharedConnection, LettuceConnectionProvider connectionProvider,
+			long timeout, int defaultDbIndex) {
 
 		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null.");
 

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

@@ -345,7 +345,11 @@ public class LettuceConnectionFactory
 			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
 		}
 
-		return new LettuceClusterConnection(connectionProvider, clusterCommandExecutor,
+		return getShareNativeConnection()
+				? new LettuceClusterConnection(
+						(StatefulRedisClusterConnection<byte[], byte[]>) getOrCreateSharedConnection().getConnection(),
+						connectionProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout())
+				: new LettuceClusterConnection(connectionProvider, clusterCommandExecutor,
 				clientConfiguration.getCommandTimeout());
 	}
 
@@ -921,7 +925,7 @@ public class LettuceConnectionFactory
 		return new StandaloneConnectionProvider((RedisClient) client, codec, readFrom);
 	}
 
-	private AbstractRedisClient createClient() {
+	protected AbstractRedisClient createClient() {
 
 		if (isRedisSentinelAware()) {
 
@@ -1065,15 +1069,10 @@ public class LettuceConnectionFactory
 		 *
 		 * @return the connection.
 		 */
-		@Nullable
 		private StatefulConnection<E, E> getNativeConnection() {
 
 			try {
 
-				if (isClusterAware() && !shareNativeClusterConnection) {
-					return null;
-				}
-
 				StatefulConnection<E, E> connection = connectionProvider.getConnection(StatefulConnection.class);
 				if (connection instanceof StatefulRedisConnection && getDatabase() > 0) {
 					((StatefulRedisConnection) connection).sync().select(getDatabase());