DATAREDIS-667 - Polishing.

Remove builder interfaces and aim for classes instead.
Also use delegation for builders avoiding constructors with too many arguments.
Alter ConnectionProvider#getConnection(Class) to return a typed connection as this is is the expected return type given the input parameters used.
Remove LettuceConnectionProvider.getConnection() method as it makes no longer sense for the non-lambda enabled interface. Remove unnecessary casts.
Add connection factory to cleanup-tracker on test class instantiation instead within the parameter supplier method to prevent shutdown by other cleanup calls that may happen in between. Reuse client resources.

Original Pull Request: #262

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

@@ -25,6 +25,7 @@ import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
  * Connection provider for Cluster connections.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
 class ClusterConnectionProvider implements LettuceConnectionProvider {
@@ -42,17 +43,16 @@ class ClusterConnectionProvider implements LettuceConnectionProvider {
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
 	 */
-	@SuppressWarnings("rawtypes")
 	@Override
-	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
-			return client.connectPubSub(codec);
+			return connectionType.cast(client.connectPubSub(codec));
 		}
 
 		if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType)
 				|| connectionType.equals(StatefulConnection.class)) {
-			return client.connect(codec);
+			return connectionType.cast(client.connect(codec));
 		}
 
 		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");

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

@@ -25,6 +25,7 @@ import java.util.Optional;
  * Default implementation of {@literal LettuceClientConfiguration}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
 class DefaultLettuceClientConfiguration implements LettuceClientConfiguration {
@@ -57,14 +58,6 @@ class DefaultLettuceClientConfiguration implements LettuceClientConfiguration {
 		return useSsl;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
-	 */
-	@Override
-	public boolean isUsePooling() {
-		return false;
-	}
-
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isVerifyPeer()
 	 */

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

@@ -19,6 +19,7 @@ import io.lettuce.core.ClientOptions;
 import io.lettuce.core.resource.ClientResources;
 
 import java.time.Duration;
+import java.util.Optional;
 
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 
@@ -26,33 +27,87 @@ import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
  * Default implementation of {@literal LettucePoolingClientConfiguration}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
-class DefaultLettucePoolingClientConfiguration extends DefaultLettuceClientConfiguration
-		implements LettucePoolingClientConfiguration {
+class DefaultLettucePoolingClientConfiguration implements LettucePoolingClientConfiguration {
 
+	private final LettuceClientConfiguration clientConfiguration;
 	private final GenericObjectPoolConfig poolConfig;
 
-	DefaultLettucePoolingClientConfiguration(boolean useSsl, boolean verifyPeer, boolean startTls,
-			ClientResources clientResources, ClientOptions clientOptions, Duration timeout, Duration shutdownTimeout,
+	DefaultLettucePoolingClientConfiguration(LettuceClientConfiguration clientConfiguration,
 			GenericObjectPoolConfig poolConfig) {
 
-		super(useSsl, verifyPeer, startTls, clientResources, clientOptions, timeout, shutdownTimeout);
-
+		this.clientConfiguration = clientConfiguration;
 		this.poolConfig = poolConfig;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
-	 */
-	@Override
-	public boolean isUsePooling() {
-		return true;
-	}
-
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration#getPoolConfig()
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUseSsl()
+	 */
+	@Override
+	public boolean isUseSsl() {
+		return clientConfiguration.isUseSsl();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isVerifyPeer()
+	 */
+	@Override
+	public boolean isVerifyPeer() {
+		return clientConfiguration.isVerifyPeer();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isStartTls()
+	 */
+	@Override
+	public boolean isStartTls() {
+		return clientConfiguration.isStartTls();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getClientResources()
+	 */
+	@Override
+	public Optional<ClientResources> getClientResources() {
+		return clientConfiguration.getClientResources();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getClientOptions()
+	 */
+	@Override
+	public Optional<ClientOptions> getClientOptions() {
+		return clientConfiguration.getClientOptions();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getCommandTimeout()
+	 */
+	@Override
+	public Duration getCommandTimeout() {
+		return clientConfiguration.getCommandTimeout();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#getShutdownTimeout()
+	 */
+	@Override
+	public Duration getShutdownTimeout() {
+		return clientConfiguration.getShutdownTimeout();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettucePoolingClientConfiguration#getPoolConfig()
 	 */
 	@Override
 	public GenericObjectPoolConfig getPoolConfig() {

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

@@ -54,11 +54,6 @@ public interface LettuceClientConfiguration {
 	 */
 	boolean isUseSsl();
 
-	/**
-	 * @return {@literal true} to use connection-pooling.
-	 */
-	boolean isUsePooling();
-
 	/**
 	 * @return {@literal true} to verify peers when using {@link #isUseSsl() SSL}.
 	 */
@@ -97,7 +92,7 @@ public interface LettuceClientConfiguration {
 	 * @return a new {@link LettuceClientConfigurationBuilder} to build {@link LettuceClientConfiguration}.
 	 */
 	static LettuceClientConfigurationBuilder builder() {
-		return new DefaultLettuceClientConfigurationBuilder();
+		return new LettuceClientConfigurationBuilder();
 	}
 
 	/**
@@ -126,101 +121,10 @@ public interface LettuceClientConfiguration {
 	}
 
 	/**
-	 * Builder for {@link LettuceClientConfiguration}.
+	 * @author Mark Paluch
+	 * @author Christoph Strobl
 	 */
-	interface LettuceClientConfigurationBuilder {
-
-		/**
-		 * Enable SSL connections.
-		 *
-		 * @return {@link LettuceSslClientConfigurationBuilder}.
-		 */
-		LettuceSslClientConfigurationBuilder useSsl();
-
-		/**
-		 * Configure {@link ClientResources}.
-		 *
-		 * @param clientResources must not be {@literal null}.
-		 * @return {@literal this} builder.
-		 * @throws IllegalArgumentException if clientResources is {@literal null}.
-		 */
-		LettuceClientConfigurationBuilder clientResources(ClientResources clientResources);
-
-		/**
-		 * Configure {@link ClientOptions}.
-		 *
-		 * @param clientOptions must not be {@literal null}.
-		 * @return {@literal this} builder.
-		 * @throws IllegalArgumentException if clientOptions is {@literal null}.
-		 */
-		LettuceClientConfigurationBuilder clientOptions(ClientOptions clientOptions);
-
-		/**
-		 * Configure a command timeout.
-		 *
-		 * @param timeout must not be {@literal null}.
-		 * @return {@literal this} builder.
-		 * @throws IllegalArgumentException if timeout is {@literal null}.
-		 */
-		LettuceClientConfigurationBuilder commandTimeout(Duration timeout);
-
-		/**
-		 * Configure a shutdown timeout.
-		 *
-		 * @param shutdownTimeout must not be {@literal null}.
-		 * @return {@literal this} builder.
-		 * @throws IllegalArgumentException if shutdownTimeout is {@literal null}.
-		 */
-		LettuceClientConfigurationBuilder shutdownTimeout(Duration shutdownTimeout);
-
-		/**
-		 * Build the {@link LettuceClientConfiguration} with the configuration applied from this builder.
-		 *
-		 * @return a new {@link LettuceClientConfiguration} object.
-		 */
-		LettuceClientConfiguration build();
-	}
-
-	/**
-	 * Builder for SSL-related {@link LettuceClientConfiguration}.
-	 */
-	interface LettuceSslClientConfigurationBuilder {
-
-		/**
-		 * Disable peer verification.
-		 *
-		 * @return {@literal this} builder.
-		 */
-		LettuceSslClientConfigurationBuilder disablePeerVerification();
-
-		/**
-		 * Enable Start TLS to send the first bytes unencrypted.
-		 *
-		 * @return {@literal this} builder.
-		 */
-		LettuceSslClientConfigurationBuilder startTls();
-
-		/**
-		 * Return to {@link LettuceClientConfigurationBuilder}.
-		 *
-		 * @return {@link LettuceClientConfigurationBuilder}.
-		 */
-		LettuceClientConfigurationBuilder and();
-
-		/**
-		 * Build the {@link LettuceClientConfiguration} with the configuration applied from this builder.
-		 *
-		 * @return a new {@link LettuceClientConfiguration} object.
-		 */
-		LettuceClientConfiguration build();
-	}
-
-	/**
-	 * Default {@link LettuceClientConfigurationBuilder} implementation to build an immutable
-	 * {@link LettuceClientConfiguration}.
-	 */
-	class DefaultLettuceClientConfigurationBuilder
-			implements LettuceClientConfigurationBuilder, LettuceSslClientConfigurationBuilder {
+	class LettuceClientConfigurationBuilder {
 
 		boolean useSsl;
 		boolean verifyPeer = true;
@@ -230,55 +134,26 @@ public interface LettuceClientConfiguration {
 		Duration timeout = Duration.ofSeconds(RedisURI.DEFAULT_TIMEOUT);
 		Duration shutdownTimeout = Duration.ofMillis(100);
 
-		DefaultLettuceClientConfigurationBuilder() {}
+		LettuceClientConfigurationBuilder() {}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#useSsl()
+		/**
+		 * Enable SSL connections.
+		 *
+		 * @return {@link LettuceSslClientConfigurationBuilder}.
 		 */
-		@Override
 		public LettuceSslClientConfigurationBuilder useSsl() {
 
 			this.useSsl = true;
-			return this;
+			return new LettuceSslClientConfigurationBuilder(this);
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#disablePeerVerification()
+		/**
+		 * Configure {@link ClientResources}.
+		 *
+		 * @param clientResources must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if clientResources is {@literal null}.
 		 */
-		@Override
-		public LettuceSslClientConfigurationBuilder disablePeerVerification() {
-
-			this.verifyPeer = false;
-			return this;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#startTls()
-		 */
-		@Override
-		public LettuceSslClientConfigurationBuilder startTls() {
-
-			this.startTls = true;
-			return this;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#and()
-		 */
-		@Override
-		public LettuceClientConfigurationBuilder and() {
-			return this;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#clientResources(io.lettuce.core.resource.ClientResources)
-		 */
-		@Override
 		public LettuceClientConfigurationBuilder clientResources(ClientResources clientResources) {
 
 			Assert.notNull(clientResources, "ClientResources must not be null!");
@@ -287,11 +162,13 @@ public interface LettuceClientConfiguration {
 			return this;
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#clientOptions(io.lettuce.core.ClientOptions)
+		/**
+		 * Configure {@link ClientOptions}.
+		 *
+		 * @param clientOptions must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if clientOptions is {@literal null}.
 		 */
-		@Override
 		public LettuceClientConfigurationBuilder clientOptions(ClientOptions clientOptions) {
 
 			Assert.notNull(clientOptions, "ClientOptions must not be null!");
@@ -300,11 +177,13 @@ public interface LettuceClientConfiguration {
 			return this;
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#timeout(java.time.Duration)
+		/**
+		 * Configure a command timeout.
+		 *
+		 * @param timeout must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if timeout is {@literal null}.
 		 */
-		@Override
 		public LettuceClientConfigurationBuilder commandTimeout(Duration timeout) {
 
 			Assert.notNull(timeout, "Duration must not be null!");
@@ -313,11 +192,13 @@ public interface LettuceClientConfiguration {
 			return this;
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#shutdownTimeout(java.time.Duration)
+		/**
+		 * Configure a shutdown timeout.
+		 *
+		 * @param shutdownTimeout must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if shutdownTimeout is {@literal null}.
 		 */
-		@Override
 		public LettuceClientConfigurationBuilder shutdownTimeout(Duration shutdownTimeout) {
 
 			Assert.notNull(timeout, "Duration must not be null!");
@@ -326,14 +207,69 @@ public interface LettuceClientConfiguration {
 			return this;
 		}
 
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#build()
+		/**
+		 * Build the {@link LettuceClientConfiguration} with the configuration applied from this builder.
+		 *
+		 * @return a new {@link LettuceClientConfiguration} object.
 		 */
-		@Override
 		public LettuceClientConfiguration build() {
+
 			return new DefaultLettuceClientConfiguration(useSsl, verifyPeer, startTls, clientResources, clientOptions,
 					timeout, shutdownTimeout);
 		}
 	}
+
+	/**
+	 * Builder for SSL-related {@link LettuceClientConfiguration}.
+	 */
+	class LettuceSslClientConfigurationBuilder {
+
+		private LettuceClientConfigurationBuilder delegate;
+
+		LettuceSslClientConfigurationBuilder(LettuceClientConfigurationBuilder delegate) {
+
+			Assert.notNull(delegate, "Delegate client configuration builder must not be null!");
+			this.delegate = delegate;
+		}
+
+		/**
+		 * Disable peer verification.
+		 *
+		 * @return {@literal this} builder.
+		 */
+		LettuceSslClientConfigurationBuilder disablePeerVerification() {
+
+			delegate.verifyPeer = false;
+			return this;
+		}
+
+		/**
+		 * Enable Start TLS to send the first bytes unencrypted.
+		 *
+		 * @return {@literal this} builder.
+		 */
+		LettuceSslClientConfigurationBuilder startTls() {
+
+			delegate.startTls = true;
+			return this;
+		}
+
+		/**
+		 * Return to {@link LettuceClientConfigurationBuilder}.
+		 *
+		 * @return {@link LettuceClientConfigurationBuilder}.
+		 */
+		LettuceClientConfigurationBuilder and() {
+			return delegate;
+		}
+
+		/**
+		 * Build the {@link LettuceClientConfiguration} with the configuration applied from this builder.
+		 *
+		 * @return a new {@link LettuceClientConfiguration} object.
+		 */
+		LettuceClientConfiguration build() {
+			return delegate.build();
+		}
+	}
 }

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

@@ -619,7 +619,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 			if (connection == null) {
 				synchronized (this) {
 					if (connection == null) {
-						this.connection = connectionProvider.getConnection();
+						this.connection = connectionProvider.getConnection(StatefulRedisClusterConnection.class);
 					}
 				}
 			}

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

@@ -820,7 +820,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	protected StatefulRedisPubSubConnection<byte[], byte[]> switchToPubSub() {
 
 		close();
-		return (StatefulRedisPubSubConnection) connectionProvider.getConnection(StatefulRedisPubSubConnection.class);
+		return connectionProvider.getConnection(StatefulRedisPubSubConnection.class);
 	}
 
 	void pipeline(LettuceResult result) {
@@ -910,7 +910,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	}
 
 	protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
-		return connectionProvider.getConnection();
+		return connectionProvider.getConnection(StatefulConnection.class);
 	}
 
 	io.lettuce.core.ScanCursor getScanCursor(long cursorId) {
@@ -1206,9 +1206,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 		private final LettucePool pool;
 
 		@Override
-		@SuppressWarnings("unchecked")
-		public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
-			return pool.getResource();
+		public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+			return connectionType.cast(pool.getResource());
 		}
 
 		@Override

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

@@ -768,7 +768,7 @@ public class LettuceConnectionFactory
 
 			StatefulRedisConnection<byte[], byte[]> connection;
 			if (!isClusterAware()) {
-				connection = connectionProvider.getConnection();
+				connection = connectionProvider.getConnection(StatefulRedisConnection.class);
 				if (getDatabase() > 0) {
 					connection.sync().select(getDatabase());
 				}
@@ -785,8 +785,9 @@ public class LettuceConnectionFactory
 
 		LettuceConnectionProvider connectionProvider = doConnectionProvider(client, codec);
 
-		if (this.clientConfiguration.isUsePooling()) {
-			return new LettucePoolingConnectionProvider(connectionProvider, this.clientConfiguration);
+		if (this.clientConfiguration instanceof LettucePoolingClientConfiguration) {
+			return new LettucePoolingConnectionProvider(connectionProvider,
+					(LettucePoolingClientConfiguration) this.clientConfiguration);
 		}
 
 		return connectionProvider;
@@ -904,14 +905,6 @@ public class LettuceConnectionFactory
 			return useSsl;
 		}
 
-		/* (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
-		 */
-		@Override
-		public boolean isUsePooling() {
-			return false;
-		}
-
 		public void setUseSsl(boolean useSsl) {
 			this.useSsl = useSsl;
 		}

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

@@ -31,6 +31,7 @@ import io.lettuce.core.api.StatefulConnection;
  * with an appropriate codec.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  * @see StatefulConnection
  */
@@ -46,18 +47,7 @@ public interface LettuceConnectionProvider {
 	 *         longer in use.
 	 */
 	@SuppressWarnings("rawtypes")
-	StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType);
-
-	/**
-	 * Request a connection.
-	 *
-	 * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no
-	 *         longer in use.
-	 */
-	@SuppressWarnings("unchecked")
-	default <T> T getConnection() {
-		return (T) getConnection(StatefulConnection.class);
-	}
+	<T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType);
 
 	/**
 	 * Release the {@link StatefulConnection connection}. Closes connection {@link StatefulConnection#close()} by default.
@@ -83,6 +73,6 @@ public interface LettuceConnectionProvider {
 		 * @return the requested connection.
 		 */
 		@SuppressWarnings("rawtypes")
-		StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType, RedisURI redisURI);
+		<T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI);
 	}
 }

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

@@ -15,32 +15,44 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import io.lettuce.core.ClientOptions;
+import io.lettuce.core.resource.ClientResources;
+
+import java.time.Duration;
+
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
-import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
 import org.springframework.util.Assert;
 
 /**
+ * Redis client configuration for lettuce using a driver level pooled connection by adding pooling specific
+ * configuration to {@link LettuceClientConfiguration}.
+ *
  * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.0
  */
 public interface LettucePoolingClientConfiguration extends LettuceClientConfiguration {
 
+	/**
+	 * @return the {@link GenericObjectPoolConfig}. Never {@literal null}.
+	 */
+	GenericObjectPoolConfig getPoolConfig();
+
 	/**
 	 * Creates a new {@link LettucePoolingClientConfigurationBuilder} to build {@link LettucePoolingClientConfiguration}
-	 * to be used with the jedis client.
+	 * to be used with the Lettuce client.
 	 *
 	 * @return a new {@link LettucePoolingClientConfigurationBuilder} to build {@link LettucePoolingClientConfiguration}.
 	 */
 	static LettucePoolingClientConfigurationBuilder builder() {
-		return new DefaultLettucePoolingClientConfigurationBuilder();
+		return new LettucePoolingClientConfigurationBuilder();
 	}
 
 	/**
-	 * Creates a default {@link LettucePoolingClientConfiguration} with:
+	 * Creates a default {@link LettucePoolingClientConfiguration} with
 	 * <dl>
 	 * <dt>SSL</dt>
 	 * <dd>no</dd>
-	 * <dt>Pooling</dt>
-	 * <dd>yes</dd>
 	 * <dt>Peer Verification</dt>
 	 * <dd>yes</dd>
 	 * <dt>Start TLS</dt>
@@ -53,6 +65,8 @@ public interface LettucePoolingClientConfiguration extends LettuceClientConfigur
 	 * <dd>60 Seconds</dd>
 	 * <dt>Shutdown Timeout</dt>
 	 * <dd>2 Seconds</dd>
+	 * <dt>pool config</dt>
+	 * <dd>default {@link GenericObjectPoolConfig}</dd>
 	 * </dl>
 	 *
 	 * @return a {@link LettucePoolingClientConfiguration} with defaults.
@@ -62,50 +76,80 @@ public interface LettucePoolingClientConfiguration extends LettuceClientConfigur
 	}
 
 	/**
-	 * @return the optional {@link GenericObjectPoolConfig}.
+	 * @author Mark Paluch
+	 * @author Christoph Strobl
 	 */
-	GenericObjectPoolConfig getPoolConfig();
-
-	/**
-	 * Builder for Pooling-related {@link LettuceClientConfiguration}.
-	 */
-	interface LettucePoolingClientConfigurationBuilder {
-
-		/**
-		 * @param poolConfig must not be {@literal null}.
-		 * @return {@literal this} builder.
-		 * @throws IllegalArgumentException if poolConfig is {@literal null}.
-		 */
-		LettucePoolingClientConfigurationBuilder poolConfig(GenericObjectPoolConfig poolConfig);
-
-		/**
-		 * Return to {@link LettuceClientConfigurationBuilder}.
-		 *
-		 * @return {@link LettuceClientConfigurationBuilder}.
-		 */
-		LettuceClientConfigurationBuilder and();
-
-		/**
-		 * Build the {@link JedisClientConfiguration} with the configuration applied from this builder.
-		 *
-		 * @return a new {@link JedisClientConfiguration} object.
-		 */
-		LettucePoolingClientConfiguration build();
-	}
-
-	/**
-	 * Default {@link LettuceClientConfigurationBuilder} implementation to build an immutable
-	 * {@link LettuceClientConfiguration}.
-	 */
-	class DefaultLettucePoolingClientConfigurationBuilder extends DefaultLettuceClientConfigurationBuilder
-			implements LettucePoolingClientConfigurationBuilder {
+	class LettucePoolingClientConfigurationBuilder extends LettuceClientConfigurationBuilder {
 
 		GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
 
+		LettucePoolingClientConfigurationBuilder() {
+			super();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#useSsl()
+		 */
 		@Override
+		public LettucePoolingSslClientConfigurationBuilder useSsl() {
+
+			super.useSsl();
+			return new LettucePoolingSslClientConfigurationBuilder(this);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#clientResources(io.lettuce.core.resource.ClientResources)
+		 */
+		@Override
+		public LettucePoolingClientConfigurationBuilder clientResources(ClientResources clientResources) {
+
+			super.clientResources(clientResources);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#clientOptions(io.lettuce.core.ClientOptions)
+		 */
+		@Override
+		public LettucePoolingClientConfigurationBuilder clientOptions(ClientOptions clientOptions) {
+
+			super.clientOptions(clientOptions);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#commandTimeout(java.time.Duration)
+		 */
+		@Override
+		public LettucePoolingClientConfigurationBuilder commandTimeout(Duration timeout) {
+
+			super.commandTimeout(timeout);
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#shutdownTimeout(java.time.Duration)
+		 */
+		@Override
+		public LettucePoolingClientConfigurationBuilder shutdownTimeout(Duration shutdownTimeout) {
+
+			super.shutdownTimeout(shutdownTimeout);
+			return this;
+		}
+
+		/**
+		 * Set the {@link GenericObjectPoolConfig} used by the driver.
+		 *
+		 * @param poolConfig must not be {@literal null}.
+		 */
 		public LettucePoolingClientConfigurationBuilder poolConfig(GenericObjectPoolConfig poolConfig) {
 
-			Assert.notNull(poolConfig, "GenericObjectPoolConfig must not be null!");
+			Assert.notNull(poolConfig, "PoolConfig must not be null!");
 
 			this.poolConfig = poolConfig;
 			return this;
@@ -117,8 +161,57 @@ public interface LettucePoolingClientConfiguration extends LettuceClientConfigur
 		 */
 		@Override
 		public LettucePoolingClientConfiguration build() {
-			return new DefaultLettucePoolingClientConfiguration(useSsl, verifyPeer, startTls, clientResources, clientOptions,
-					timeout, shutdownTimeout, poolConfig);
+			return new DefaultLettucePoolingClientConfiguration(super.build(), poolConfig);
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 */
+	class LettucePoolingSslClientConfigurationBuilder extends LettuceSslClientConfigurationBuilder {
+
+		LettucePoolingSslClientConfigurationBuilder(LettucePoolingClientConfigurationBuilder delegate) {
+			super(delegate);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#and()
+		 */
+		@Override
+		LettucePoolingClientConfigurationBuilder and() {
+			return (LettucePoolingClientConfigurationBuilder) super.and();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#disablePeerVerification()
+		 */
+		@Override
+		LettucePoolingSslClientConfigurationBuilder disablePeerVerification() {
+
+			super.disablePeerVerification();
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#startTls()
+		 */
+		@Override
+		LettucePoolingSslClientConfigurationBuilder startTls() {
+
+			super.startTls();
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceSslClientConfigurationBuilder#build()
+		 */
+		@Override
+		LettucePoolingClientConfiguration build() {
+			return (LettucePoolingClientConfiguration) super.build();
 		}
 	}
 }

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

@@ -27,6 +27,7 @@ import org.apache.commons.pool2.impl.GenericObjectPool;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.data.redis.connection.PoolException;
+import org.springframework.util.Assert;
 
 /**
  * {@link LettuceConnectionProvider} with connection pooling support. This connection provider holds multiple pools (one
@@ -37,6 +38,7 @@ import org.springframework.data.redis.connection.PoolException;
  * to close the pools.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  * @see #getConnection(Class)
  */
@@ -51,10 +53,13 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Dis
 	private final Map<Class<?>, GenericObjectPool<StatefulConnection<?, ?>>> pools = new ConcurrentHashMap<>(32);
 
 	LettucePoolingConnectionProvider(LettuceConnectionProvider connectionProvider,
-			LettuceClientConfiguration clientConfiguration) {
+			LettucePoolingClientConfiguration clientConfiguration) {
+
+		Assert.notNull(connectionProvider, "ConnectionProvider must not be null!");
+		Assert.notNull(clientConfiguration, "ClientConfiguration must not be null!");
 
 		this.connectionProvider = connectionProvider;
-		this.poolConfig = ((LettucePoolingClientConfiguration) clientConfiguration).getPoolConfig();
+		this.poolConfig = clientConfiguration.getPoolConfig();
 	}
 
 	/*
@@ -62,8 +67,7 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Dis
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
 	 */
 	@Override
-	@SuppressWarnings("rawtypes")
-	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
 
 		GenericObjectPool<StatefulConnection<?, ?>> pool = pools.computeIfAbsent(connectionType, poolType -> {
 			return ConnectionPoolSupport.createGenericObjectPool(() -> connectionProvider.getConnection(connectionType),
@@ -71,11 +75,12 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Dis
 		});
 
 		try {
+
 			StatefulConnection<?, ?> connection = pool.borrowObject();
 
 			poolRef.put(connection, pool);
 
-			return connection;
+			return connectionType.cast(connection);
 		} catch (Exception e) {
 			throw new PoolException("Could not get a resource from the pool", e);
 		}

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

@@ -21,7 +21,6 @@ import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.codec.RedisCodec;
-import org.springframework.data.redis.connection.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -31,6 +30,7 @@ import java.util.function.Function;
 import org.reactivestreams.Publisher;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.data.redis.connection.*;
 import org.springframework.util.Assert;
 
 /**
@@ -58,7 +58,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
 		this.connectionProvider = connectionProvider;
-		this.connection = connectionProvider.getConnection();
+		this.connection = connectionProvider.getConnection(StatefulConnection.class);
 	}
 
 	/*

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

@@ -94,7 +94,12 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	public LettuceSentinelConnection(RedisClient redisClient) {
 
 		Assert.notNull(redisClient, "Cannot create LettuceSentinelConnection using 'null' as client.");
-		this.provider = t -> redisClient.connectSentinel();
+		this.provider = new LettuceConnectionProvider() {
+			@Override
+			public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> t) {
+				return t.cast(redisClient.connectSentinel());
+			}
+		};
 		init();
 	}
 
@@ -106,7 +111,12 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	protected LettuceSentinelConnection(StatefulRedisSentinelConnection<String, String> connection) {
 
 		Assert.notNull(connection, "Cannot create LettuceSentinelConnection using 'null' as connection.");
-		this.provider = t -> connection;
+		this.provider = new LettuceConnectionProvider() {
+			@Override
+			public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> t) {
+				return t.cast(connection);
+			}
+		};
 		init();
 	}
 
@@ -224,7 +234,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	private void init() {
 
 		if (connection == null) {
-			connection = (StatefulRedisSentinelConnection) provider.getConnection(StatefulRedisSentinelConnection.class);
+			connection = provider.getConnection(StatefulRedisSentinelConnection.class);
 		}
 	}
 
@@ -263,10 +273,9 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 		 * (non-Javadoc)
 		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
 		 */
-		@SuppressWarnings("rawtypes")
 		@Override
-		public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
-			return redisClient.connectSentinel();
+		public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+			return connectionType.cast(redisClient.connectSentinel());
 		}
 
 		/*

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

@@ -26,6 +26,7 @@ import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvid
 
 /**
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
 class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetAware {
@@ -43,20 +44,18 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
 	 */
-	@SuppressWarnings("rawtypes")
 	@Override
-	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
-
+	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
 		if (connectionType.equals(StatefulRedisSentinelConnection.class)) {
-			return client.connectSentinel();
+			return connectionType.cast(client.connectSentinel());
 		}
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
-			return client.connectPubSub(codec);
+			return connectionType.cast(client.connectPubSub(codec));
 		}
 
 		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
-			return client.connect(codec);
+			return connectionType.cast(client.connect(codec));
 		}
 
 		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
@@ -66,20 +65,19 @@ class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetA
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnection(java.lang.Class, io.lettuce.core.RedisURI)
 	 */
-	@SuppressWarnings("rawtypes")
 	@Override
-	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType, RedisURI redisURI) {
+	public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI) {
 
 		if (connectionType.equals(StatefulRedisSentinelConnection.class)) {
-			return client.connectSentinel(redisURI);
+			return connectionType.cast(client.connectSentinel(redisURI));
 		}
 
 		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
-			return client.connectPubSub(codec, redisURI);
+			return connectionType.cast(client.connectPubSub(codec, redisURI));
 		}
 
 		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
-			return client.connect(codec, redisURI);
+			return connectionType.cast(client.connect(codec, redisURI));
 		}
 
 		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");