DATAREDIS-721 - Switch to non-blocking connect methods.

We now connect asynchronously to expose non-blocking connection behavior for reactive API usage. We also use a non-blocking, asynchronous pool implementation for connections that are used through a reactive API. Shared connection usage, which is blocking on the very first access can be pre-initialized during connection factory initialization instead of initialization on first access:

LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory();
connectionFactory.setEagerInitialization(true);
connectionFactory.afterPropertiesSet();

We no longer require offloading of blocking connects to a dedicated scheduler which makes reactive API usage fully non-blocking.

Original Pull Request: #315

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

@@ -23,6 +23,7 @@ import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -40,6 +41,10 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien
 	private final RedisCodec<?, ?> codec;
 	private final Optional<ReadFrom> readFrom;
 
+	private final Object monitor = new Object();
+
+	private volatile boolean initialized;
+
 	/**
 	 * Create new {@link ClusterConnectionProvider}.
 	 *
@@ -70,25 +75,39 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
 	 */
 	@Override
-	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+	public <T extends StatefulConnection<?, ?>> CompletableFuture<T> getConnectionAsync(Class<T> connectionType) {
+
+		if (!initialized) {
+
+			// partitions have to be initialized before asynchronous usage.
+			// Needs to happen only once. Initialize eagerly if
+			// blocking is not an options.
+			synchronized (monitor) {
+				if (!initialized) {
+					client.getPartitions();
+					initialized = true;
+				}
+			}
+		}
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
-			return connectionType.cast(client.connectPubSub(codec));
+			return client.connectPubSubAsync(codec).thenApply(connectionType::cast);
 		}
 
 		if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType)
 				|| connectionType.equals(StatefulConnection.class)) {
 
-			StatefulRedisClusterConnection<?, ?> connection = client.connect(codec);
-			readFrom.ifPresent(connection::setReadFrom);
-
-			return connectionType.cast(connection);
+			return client.connectAsync(codec).thenApply(conn -> {
+				readFrom.ifPresent(conn::setReadFrom);
+				return connectionType.cast(conn);
+			});
 		}
 
-		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+		return LettuceFutureUtils
+				.failed(new UnsupportedOperationException("Connection type " + connectionType + " not supported!"));
 	}
 
 	/*

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

@@ -49,6 +49,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -1253,11 +1254,28 @@ public class LettuceConnection extends AbstractRedisConnection {
 
 		private final LettucePool pool;
 
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+		 */
 		@Override
 		public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
 			return connectionType.cast(pool.getResource());
 		}
 
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
+		 */
+		@Override
+		public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType) {
+			throw new UnsupportedOperationException("Async operations not supported!");
+		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection)
+		 */
 		@Override
 		@SuppressWarnings("unchecked")
 		public void release(StatefulConnection<?, ?> connection) {

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

@@ -101,6 +101,7 @@ public class LettuceConnectionFactory
 	private @Nullable LettuceConnectionProvider reactiveConnectionProvider;
 	private boolean validateConnection = false;
 	private boolean shareNativeConnection = true;
+	private boolean eagerInitialization = false;
 	private @Nullable SharedConnection<byte[]> connection;
 	private @Nullable SharedConnection<ByteBuffer> reactiveConnection;
 	private @Nullable LettucePool pool;
@@ -280,6 +281,10 @@ public class LettuceConnectionFactory
 					new LettuceClusterConnection.LettuceClusterNodeResourceProvider(this.connectionProvider),
 					EXCEPTION_TRANSLATION);
 		}
+
+		if (getEagerInitialization() && getShareNativeConnection()) {
+			initConnection();
+		}
 	}
 
 	/*
@@ -629,6 +634,30 @@ public class LettuceConnectionFactory
 		this.shareNativeConnection = shareNativeConnection;
 	}
 
+	/**
+	 * Indicates {@link #setShareNativeConnection(boolean) shared connections} should be eagerly initialized. Eager
+	 * initialization requires a running Redis instance during application startup to allow early validation of connection
+	 * factory configuration. Eager initialization also prevents blocking connect while using reactive API and is
+	 * recommended for reactive API usage.
+	 *
+	 * @return {@link true} if the shared connection is initialized upon {@link #afterPropertiesSet()}.
+	 * @since 2.2
+	 */
+	public boolean getEagerInitialization() {
+		return eagerInitialization;
+	}
+
+	/**
+	 * Enables eager initialization of {@link #setShareNativeConnection(boolean) shared connections}.
+	 *
+	 * @param eagerInitialization enable eager connection shared connection initialization upon
+	 *          {@link #afterPropertiesSet()}.
+	 * @since 2.2
+	 */
+	public void setEagerInitialization(boolean eagerInitialization) {
+		this.eagerInitialization = eagerInitialization;
+	}
+
 	/**
 	 * Returns the index of the database.
 	 *

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

@@ -18,6 +18,9 @@ package org.springframework.data.redis.connection.lettuce;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulConnection;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
 /**
  * Defines a provider for Lettuce connections.
  * <p />
@@ -46,7 +49,20 @@ public interface LettuceConnectionProvider {
 	 * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no
 	 *         longer in use.
 	 */
-	<T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType);
+	default <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+		return LettuceFutureUtils.join(getConnectionAsync(connectionType));
+	}
+
+	/**
+	 * Request asynchronously a connection given {@code connectionType}. Providing a connection type allows specialization
+	 * to provide a more specific connection type.
+	 *
+	 * @param connectionType must not be {@literal null}.
+	 * @return a {@link CompletionStage} that is notified with the connection progress. Must be
+	 *         {@link #releaseAsync(StatefulConnection) released} if the connection is no longer in use.
+	 * @since 2.2
+	 */
+	<T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType);
 
 	/**
 	 * Release the {@link StatefulConnection connection}. Closes connection {@link StatefulConnection#close()} by default.
@@ -55,12 +71,26 @@ public interface LettuceConnectionProvider {
 	 * @param connection must not be {@literal null}.
 	 */
 	default void release(StatefulConnection<?, ?> connection) {
-		connection.close();
+		LettuceFutureUtils.join(releaseAsync(connection));
+	}
+
+	/**
+	 * Release asynchronously the {@link StatefulConnection connection}. Closes connection
+	 * {@link StatefulConnection#closeAsync()} by default. Implementations may choose whether they override this method
+	 * and return the connection to a pool.
+	 *
+	 * @param connection must not be {@literal null}.
+	 * @return Close {@link CompletableFuture future} notified once the connection is released.
+	 * @since 2.2
+	 */
+	default CompletableFuture<Void> releaseAsync(StatefulConnection<?, ?> connection) {
+		return connection.closeAsync();
 	}
 
 	/**
 	 * Extension to {@link LettuceConnectionProvider} for providers that allow connection creation to specific nodes.
 	 */
+	@FunctionalInterface
 	interface TargetAware {
 
 		/**
@@ -71,6 +101,20 @@ public interface LettuceConnectionProvider {
 		 * @param redisURI must not be {@literal null}.
 		 * @return the requested connection.
 		 */
-		<T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI);
+		default <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI) {
+			return LettuceFutureUtils.join(getConnectionAsync(connectionType, redisURI));
+		}
+
+		/**
+		 * Request asynchronously a connection given {@code connectionType} for a specific {@link RedisURI}. Providing a
+		 * connection type allows specialization to provide a more specific connection type.
+		 *
+		 * @param connectionType must not be {@literal null}.
+		 * @param redisURI must not be {@literal null}.
+		 * @return a {@link CompletionStage} that is notified with the connection progress.
+		 * @since 2.2
+		 */
+		<T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType,
+				RedisURI redisURI);
 	}
 }

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection.lettuce;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Utility methods to interact with {@link CompletableFuture} and {@link CompletionStage}.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+class LettuceFutureUtils {
+
+	/**
+	 * Creates a {@link CompletableFuture} that is completed {@link CompletableFuture#exceptionally(Function)
+	 * exceptionally} given {@link Throwable}. This utility method allows exceptionally future creation with a single
+	 * invocation.
+	 *
+	 * @param t must not be {@literal null}.
+	 * @return the completed {@link CompletableFuture future}.
+	 */
+	static <T> CompletableFuture<T> failed(Throwable t) {
+
+		Assert.notNull(t, "Throwable must not be null!");
+
+		CompletableFuture<T> future = new CompletableFuture<>();
+		future.completeExceptionally(t);
+
+		return future;
+	}
+
+	/**
+	 * Synchronizes a {@link CompletableFuture} result by {@link CompletableFuture#join() waiting until the future is
+	 * complete}. This method preserves {@link RuntimeException}s that may get thrown as result of future completion.
+	 * Checked exceptions are thrown encapsulated within {@link java.util.concurrent.CompletionException}.
+	 *
+	 * @param future must not be {@literal null}.
+	 * @throws RuntimeException thrown if the future is completed with a {@link RuntimeException}.
+	 * @throws CompletionException thrown if the future is completed with a checked exception.
+	 * @return the future result if completed normally.
+	 */
+	@Nullable
+	static <T> T join(CompletionStage<T> future) throws RuntimeException, CompletionException {
+
+		Assert.notNull(future, "CompletableFuture must not be null!");
+
+		try {
+			return future.toCompletableFuture().join();
+		} catch (Exception e) {
+
+			Throwable exceptionToUse = e;
+
+			if (e instanceof CompletionException) {
+				exceptionToUse = new LettuceExceptionConverter().convert((Exception) e.getCause());
+				if (exceptionToUse == null) {
+					exceptionToUse = e.getCause();
+				}
+			}
+
+			if (exceptionToUse instanceof RuntimeException) {
+				throw (RuntimeException) exceptionToUse;
+			}
+
+			throw new CompletionException(exceptionToUse);
+		}
+	}
+
+	/**
+	 * Returns a {@link Function} that ignores {@link CompletionStage#exceptionally(Function) exceptional completion} by
+	 * recovering to {@code null}. This allows to progress with a previously failed {@link CompletionStage} without regard
+	 * to the actual success/exception state.
+	 *
+	 * @return
+	 */
+	static <T> Function<Throwable, T> ignoreErrors() {
+		return ignored -> null;
+	}
+}

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

@@ -17,9 +17,17 @@ package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.AbstractRedisClient;
 import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.support.AsyncConnectionPoolSupport;
+import io.lettuce.core.support.AsyncPool;
+import io.lettuce.core.support.BoundedPoolConfig;
+import io.lettuce.core.support.CommonsPool2ConfigConverter;
 import io.lettuce.core.support.ConnectionPoolSupport;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.commons.logging.Log;
@@ -32,11 +40,17 @@ import org.springframework.util.Assert;
 
 /**
  * {@link LettuceConnectionProvider} with connection pooling support. This connection provider holds multiple pools (one
- * per connection type) for contextualized connection allocation.
+ * per connection type and allocation type (synchronous/asynchronous)) for contextualized connection allocation.
  * <p />
  * Each allocated connection is tracked and to be returned into the pool which created the connection. Instances of this
  * class require {@link #destroy() disposal} to de-allocate lingering connections that were not returned to the pool and
  * to close the pools.
+ * <p />
+ * This provider maintains separate pools due to the allocation nature (synchronous/asynchronous). Asynchronous
+ * connection pooling requires a non-blocking allocation API. Connections requested asynchronously can be returned
+ * synchronously and vice versa. A connection obtained synchronously is returned to the synchronous pool even if
+ * {@link #releaseAsync(StatefulConnection) released asynchronously}. This is an undesired case as the synchronous pool
+ * will block the asynchronous flow for the time of release.
  *
  * @author Mark Paluch
  * @author Christoph Strobl
@@ -51,7 +65,14 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
 	private final GenericObjectPoolConfig poolConfig;
 	private final Map<StatefulConnection<?, ?>, GenericObjectPool<StatefulConnection<?, ?>>> poolRef = new ConcurrentHashMap<>(
 			32);
+
+	private final Map<StatefulConnection<?, ?>, AsyncPool<StatefulConnection<?, ?>>> asyncPoolRef = new ConcurrentHashMap<>(
+			32);
+	private final Map<CompletableFuture<StatefulConnection<?, ?>>, AsyncPool<StatefulConnection<?, ?>>> inProgressAsyncPoolRef = new ConcurrentHashMap<>(
+			32);
 	private final Map<Class<?>, GenericObjectPool<StatefulConnection<?, ?>>> pools = new ConcurrentHashMap<>(32);
+	private final Map<Class<?>, AsyncPool<StatefulConnection<?, ?>>> asyncPools = new ConcurrentHashMap<>(32);
+	private final BoundedPoolConfig asyncPoolConfig;
 
 	LettucePoolingConnectionProvider(LettuceConnectionProvider connectionProvider,
 			LettucePoolingClientConfiguration clientConfiguration) {
@@ -61,6 +82,7 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
 
 		this.connectionProvider = connectionProvider;
 		this.poolConfig = clientConfiguration.getPoolConfig();
+		this.asyncPoolConfig = CommonsPool2ConfigConverter.bounded(this.poolConfig);
 	}
 
 	/*
@@ -87,6 +109,33 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
 		}
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
+	 */
+	@Override
+	public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType) {
+
+		AsyncPool<StatefulConnection<?, ?>> pool = asyncPools.computeIfAbsent(connectionType, poolType -> {
+
+			return AsyncConnectionPoolSupport.createBoundedObjectPool(
+					() -> connectionProvider.getConnectionAsync(connectionType).thenApply(connectionType::cast), asyncPoolConfig,
+					false);
+		});
+
+		CompletableFuture<StatefulConnection<?, ?>> acquire = pool.acquire();
+
+		inProgressAsyncPoolRef.put(acquire, pool);
+		return acquire.whenComplete((conn, e) -> {
+
+			inProgressAsyncPoolRef.remove(acquire);
+
+			if (conn != null) {
+				asyncPoolRef.put(conn, pool);
+			}
+		}).thenApply(connectionType::cast);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.RedisClientProvider#getRedisClient()
@@ -113,13 +162,47 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
 		GenericObjectPool<StatefulConnection<?, ?>> pool = poolRef.remove(connection);
 
 		if (pool == null) {
-			throw new PoolException("Returned connection " + connection
-					+ " was either previously returned or does not belong to this connection provider");
+
+			AsyncPool<StatefulConnection<?, ?>> asyncPool = asyncPoolRef.remove(connection);
+
+			if (asyncPool == null) {
+				throw new PoolException("Returned connection " + connection
+						+ " was either previously returned or does not belong to this connection provider");
+			}
+
+			asyncPool.release(connection).join();
+			return;
 		}
 
 		pool.returnObject(connection);
 	}
 
+	/* 
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#releaseAsync(io.lettuce.core.api.StatefulConnection)
+	 */
+	@Override
+	public CompletableFuture<Void> releaseAsync(StatefulConnection<?, ?> connection) {
+
+		GenericObjectPool<StatefulConnection<?, ?>> blockingPool = poolRef.remove(connection);
+
+		if (blockingPool != null) {
+
+			log.warn("Releasing asynchronously a connection that was obtained from a non-blocking pool");
+			blockingPool.returnObject(connection);
+			return CompletableFuture.completedFuture(null);
+		}
+
+		AsyncPool<StatefulConnection<?, ?>> pool = asyncPoolRef.remove(connection);
+
+		if (pool == null) {
+			return LettuceFutureUtils.failed(new PoolException("Returned connection " + connection
+					+ " was either previously returned or does not belong to this connection provider"));
+		}
+
+		return pool.release(connection);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
@@ -127,15 +210,47 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red
 	@Override
 	public void destroy() throws Exception {
 
-		if (!poolRef.isEmpty()) {
-
+		List<CompletableFuture<?>> futures = new ArrayList<>();
+		if (!poolRef.isEmpty() || !asyncPoolRef.isEmpty()) {
 			log.warn("LettucePoolingConnectionProvider contains unreleased connections");
+		}
+
+		if (!inProgressAsyncPoolRef.isEmpty()) {
+
+			log.warn("LettucePoolingConnectionProvider has active connection retrievals");
+			inProgressAsyncPoolRef.forEach((k, v) -> futures.add(k.thenApply(StatefulConnection::closeAsync)));
+		}
+
+		if (!poolRef.isEmpty()) {
 
 			poolRef.forEach((connection, pool) -> pool.returnObject(connection));
 			poolRef.clear();
 		}
 
+		if (!asyncPoolRef.isEmpty()) {
+
+			asyncPoolRef.forEach((connection, pool) -> futures.add(pool.release(connection)));
+			asyncPoolRef.clear();
+		}
+
 		pools.forEach((type, pool) -> pool.close());
+
+		CompletableFuture
+				.allOf(futures.stream().map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors()))
+						.toArray(CompletableFuture[]::new)) //
+				.thenCompose(ignored -> {
+
+					CompletableFuture[] poolClose = asyncPools.values().stream().map(AsyncPool::closeAsync)
+							.map(it -> it.exceptionally(LettuceFutureUtils.ignoreErrors())).toArray(CompletableFuture[]::new);
+
+					return CompletableFuture.allOf(poolClose);
+				}) //
+				.thenRun(() -> {
+					asyncPoolRef.clear();
+					inProgressAsyncPoolRef.clear();
+				}) //
+				.join();
+
 		pools.clear();
 	}
 }

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
@@ -240,7 +241,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 					.map(it -> it.getConnection(node.getId()).reactive());
 		}
 
-		return getConnection().cast(StatefulRedisClusterConnection.class)
-				.map(it -> it.getConnection(node.getHost(), node.getPort()).reactive());
+		return getConnection().flatMap(it -> Mono.fromCompletionStage(it.getConnectionAsync(node.getHost(), node.getPort()))
+				.map(StatefulRedisConnection::reactive));
 	}
 }

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import static org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection.AsyncConnect.State.*;
+
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
@@ -24,10 +26,8 @@ import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 import java.nio.ByteBuffer;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
@@ -237,7 +237,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#closeLater()
 	 */
 	public Mono<Void> closeLater() {
-		return Mono.fromRunnable(() -> dedicatedConnection.close());
+		return Mono.fromRunnable(dedicatedConnection::close);
 	}
 
 	protected Mono<? extends StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {
@@ -335,8 +335,9 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * {@link #close()} methods:
 	 * <ol>
 	 * <li>Initial state: Not connected. Calling {@link #getConnection()} will transition to the next state.</li>
-	 * <li>Connection requested: First caller to {@link #getConnection()} initiates an asynchronous connect. Connection is
-	 * accessible through the resulting {@link Mono}.</li>
+	 * <li>Connection requested: First subscriber to {@link #getConnection()} initiates an asynchronous connect.
+	 * Connection is accessible through the resulting {@link Mono} and will be cached. Connection will be closed if
+	 * {@link AsyncConnect} gets closed before the connection publisher emits the actual connection.</li>
 	 * <li>Closing: Call to {@link #close()} initiates connection closing.</li>
 	 * <li>Closed: Connection is closed.</li>
 	 * </ol>
@@ -345,24 +346,42 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * @author Christoph Strobl
 	 * @since 2.0.1
 	 */
-	static class AsyncConnect<T extends StatefulConnection<?, ?>> {
+	static class AsyncConnect<T extends io.lettuce.core.api.StatefulConnection<?, ?>> {
 
 		private final Mono<T> connectionPublisher;
 		private final LettuceConnectionProvider connectionProvider;
 
 		private AtomicReference<State> state = new AtomicReference<>(State.INITIAL);
-		private volatile @Nullable CompletableFuture<T> connection;
+		private volatile @Nullable StatefulConnection<ByteBuffer, ByteBuffer> connection;
 
 		@SuppressWarnings("unchecked")
-		AsyncConnect(LettuceConnectionProvider connectionProvider, Class<? extends T> connectionType) {
+		AsyncConnect(LettuceConnectionProvider connectionProvider, Class<T> connectionType) {
 
 			Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
+			Assert.notNull(connectionType, "Connection type must not be null!");
 
 			this.connectionProvider = connectionProvider;
 
-			Mono<T> defer = Mono.defer(() -> Mono.<T> just(connectionProvider.getConnection(connectionType)));
+			Mono<StatefulConnection> defer = Mono
+					.defer(() -> Mono.fromCompletionStage(connectionProvider.getConnectionAsync(connectionType)));
 
-			this.connectionPublisher = defer.subscribeOn(Schedulers.elastic());
+			this.connectionPublisher = defer.doOnNext(it -> {
+
+				if (isClosing(this.state.get())) {
+					it.closeAsync();
+				} else {
+					connection = it;
+				}
+			}) //
+					.cache() //
+					.handle((connection, sink) -> {
+
+						if (isClosing(this.state.get())) {
+							sink.error(new IllegalStateException("Unable to connect. Connection is closed!"));
+						} else {
+							sink.next((T) connection);
+						}
+					});
 		}
 
 		/**
@@ -373,33 +392,14 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		 */
 		Mono<T> getConnection() {
 
-			if (state.get() == State.CLOSED) {
-				throw new IllegalStateException("Unable to connect. Connection is closed!");
+			State state = this.state.get();
+			if (isClosing(state)) {
+				return Mono.error(new IllegalStateException("Unable to connect. Connection is closed!"));
 			}
 
-			CompletableFuture<T> connection = this.connection;
+			this.state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED);
 
-			if (connection != null) {
-				return Mono.fromCompletionStage(connection);
-			}
-
-			if (state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED)) {
-				this.connection = connectionPublisher.toFuture();
-			}
-
-			while (true) {
-
-				connection = this.connection;
-				if (connection != null) {
-					return Mono.fromCompletionStage(connection);
-				}
-
-				Thread thread = Thread.currentThread();
-				if (thread.isInterrupted()) {
-					thread.interrupt();
-					return Mono.error(new InterruptedException());
-				}
-			}
+			return connectionPublisher;
 		}
 
 		/**
@@ -407,11 +407,21 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		 */
 		void close() {
 
-			if (state.compareAndSet(State.CONNECTION_REQUESTED, State.CLOSING)) {
-				getConnection().doOnSuccess(connectionProvider::release).doOnSuccess(it -> state.set(State.CLOSED)).block();
-			}
+			if (state.compareAndSet(State.INITIAL, CLOSING) || state.compareAndSet(State.CONNECTION_REQUESTED, CLOSING)) {
 
-			state.compareAndSet(State.INITIAL, State.CLOSED);
+				StatefulConnection<ByteBuffer, ByteBuffer> connection = this.connection;
+				this.connection = null;
+
+				if (connection != null) {
+					LettuceFutureUtils.join(connectionProvider.releaseAsync(connection));
+				}
+
+				state.set(State.CLOSED);
+			}
+		}
+
+		private static boolean isClosing(State state) {
+			return state == State.CLOSING || state == State.CLOSED;
 		}
 
 		enum State {

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

@@ -16,14 +16,17 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
 import io.lettuce.core.RedisURI.Builder;
 import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.codec.StringCodec;
 import io.lettuce.core.resource.ClientResources;
 import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
 import io.lettuce.core.sentinel.api.sync.RedisSentinelCommands;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
@@ -99,6 +102,11 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 			public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> t) {
 				return t.cast(redisClient.connectSentinel());
 			}
+
+			@Override
+			public <T extends StatefulConnection<?, ?>> CompletableFuture<T> getConnectionAsync(Class<T> t) {
+				return CompletableFuture.completedFuture(t.cast(connection));
+			}
 		};
 		init();
 	}
@@ -116,6 +124,11 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 			public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> t) {
 				return t.cast(connection);
 			}
+
+			@Override
+			public <T extends StatefulConnection<?, ?>> CompletableFuture<T> getConnectionAsync(Class<T> t) {
+				return CompletableFuture.completedFuture(t.cast(connection));
+			}
 		};
 		init();
 	}
@@ -253,12 +266,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	private static class DedicatedClientConnectionProvider implements LettuceConnectionProvider {
 
 		private final RedisClient redisClient;
+		private final RedisURI uri;
 
 		DedicatedClientConnectionProvider(String host, int port) {
 
 			Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
 
-			redisClient = RedisClient.create(Builder.redis(host, port).build());
+			uri = Builder.redis(host, port).build();
+			redisClient = RedisClient.create(uri);
 		}
 
 		DedicatedClientConnectionProvider(String host, int port, ClientResources clientResources) {
@@ -266,16 +281,17 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 			Assert.notNull(clientResources, "Cannot create LettuceSentinelConnection using 'null' as ClientResources.");
 			Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
 
-			redisClient = RedisClient.create(clientResources, Builder.redis(host, port).build());
+			this.uri = Builder.redis(host, port).build();
+			redisClient = RedisClient.create(clientResources, uri);
 		}
 
-		/*
+		/* 
 		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
 		 */
 		@Override
-		public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
-			return connectionType.cast(redisClient.connectSentinel());
+		public <T extends StatefulConnection<?, ?>> CompletableFuture<T> getConnectionAsync(Class<T> connectionType) {
+			return redisClient.connectSentinelAsync(StringCodec.UTF8, uri).thenApply(connectionType::cast);
 		}
 
 		/*
@@ -288,5 +304,15 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 			connection.close();
 			redisClient.shutdown();
 		}
+
+		/* 
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#releaseAsync(io.lettuce.core.api.StatefulConnection)
+		 */
+		@Override
+		public CompletableFuture<Void> releaseAsync(StatefulConnection<?, ?> connection) {
+			return connection.closeAsync().exceptionally(LettuceFutureUtils.ignoreErrors())
+					.thenCompose(it -> redisClient.shutdownAsync());
+		}
 	}
 }

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

@@ -21,12 +21,15 @@ import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.codec.StringCodec;
 import io.lettuce.core.masterslave.MasterSlave;
 import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
 
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
@@ -74,7 +77,7 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 
 		redisURISupplier = new Supplier<RedisURI>() {
 
-			AtomicReference<RedisURI> uriFieldReference = new AtomicReference();
+			AtomicReference<RedisURI> uriFieldReference = new AtomicReference<>();
 
 			@Override
 			public RedisURI get() {
@@ -95,7 +98,6 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
 	 */
-	@SuppressWarnings("null")
 	@Override
 	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
 
@@ -118,27 +120,38 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnection(java.lang.Class, io.lettuce.core.RedisURI)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
 	 */
-	@SuppressWarnings("null")
 	@Override
-	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI) {
+	public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType) {
+		return getConnectionAsync(connectionType, redisURISupplier.get());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnectionAsync(java.lang.Class, io.lettuce.core.RedisURI)
+	 */
+	@SuppressWarnings({ "null", "unchecked", "rawtypes" })
+	@Override
+	public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType,
+			RedisURI redisURI) {
 
 		if (connectionType.equals(StatefulRedisSentinelConnection.class)) {
-			return connectionType.cast(client.connectSentinel(redisURI));
+			return client.connectSentinelAsync(StringCodec.UTF8, redisURI).thenApply(connectionType::cast);
 		}
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
-			return connectionType.cast(client.connectPubSub(codec, redisURI));
+			return client.connectPubSubAsync(codec, redisURI).thenApply(connectionType::cast);
 		}
 
 		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
-
-			return connectionType
-					.cast(readFrom.map(it -> this.masterReplicaConnection(redisURI, it)).orElseGet(() -> client.connect(codec)));
+			return readFrom.map(it -> this.masterReplicaConnectionAsync(redisURI, it)) //
+					.orElseGet(() -> (CompletionStage) client.connectAsync(codec, redisURI)) //
+					.thenApply(connectionType::cast);
 		}
 
-		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+		return LettuceFutureUtils
+				.failed(new UnsupportedOperationException("Connection type " + connectionType + " not supported!"));
 	}
 
 	private StatefulRedisConnection masterReplicaConnection(RedisURI redisUri, ReadFrom readFrom) {
@@ -150,4 +163,17 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 		return connection;
 	}
 
+	private CompletionStage<StatefulRedisConnection<?, ?>> masterReplicaConnectionAsync(RedisURI redisUri,
+			ReadFrom readFrom) {
+
+		CompletableFuture<? extends StatefulRedisMasterSlaveConnection<?, ?>> connection = MasterSlave.connectAsync(client,
+				codec, redisUri);
+
+		return connection.thenApply(conn -> {
+
+			conn.setReadFrom(readFrom);
+
+			return conn;
+		});
+	}
 }

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

@@ -25,6 +25,8 @@ import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection;
 
 import java.util.Collection;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 import org.springframework.lang.Nullable;
 
@@ -79,4 +81,27 @@ class StaticMasterReplicaConnectionProvider implements LettuceConnectionProvider
 
 		throw new UnsupportedOperationException(String.format("Connection type %s not supported!", connectionType));
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class)
+	 */
+	@Override
+	public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType) {
+
+		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
+
+			// See https://github.com/lettuce-io/lettuce-core/issues/845 for MasterSlave -> MasterReplica change.
+			CompletableFuture<? extends StatefulRedisMasterSlaveConnection<?, ?>> connection = MasterSlave
+					.connectAsync(client, codec, nodes);
+
+			connection.thenApply(it -> {
+
+				readFrom.ifPresent(readFrom -> it.setReadFrom(readFrom));
+				return connectionType.cast(connection);
+			});
+		}
+
+		throw new UnsupportedOperationException(String.format("Connection type %s not supported!", connectionType));
+	}
 }