DATAREDIS-1189 - Consistently translate Lettuce connection exceptions.

We now translate consistently connection/pooling exceptions in LettuceConnectionFactory by wrapping LettuceConnectionProvider with a variant that considers exception translation.

Author: mp911de

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

@@ -688,11 +688,7 @@ public RedisClusterCommands<byte[], byte[]> getResourceForSpecificNode(RedisClus
 }
 }
 
-try {
-return connection.getConnection(node.getHost(), node.getPort()).sync();
-} catch (RedisException e) {
-throw new DataAccessResourceFailureException(e.getMessage(), e);
-}
+return connection.getConnection(node.getHost(), node.getPort()).sync();
 }
 
 @Override

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

@@ -21,7 +21,7 @@
 import io.lettuce.core.ClientOptions;
 import io.lettuce.core.ReadFrom;
 import io.lettuce.core.RedisClient;
-import io.lettuce.core.RedisException;
+import io.lettuce.core.RedisConnectionException;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
@@ -36,12 +36,15 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
@@ -54,6 +57,7 @@
 import org.springframework.data.redis.connection.RedisConfiguration.DomainSocketConfiguration;
 import org.springframework.data.redis.connection.RedisConfiguration.WithDatabaseIndex;
 import org.springframework.data.redis.connection.RedisConfiguration.WithPassword;
+import org.springframework.data.redis.connection.lettuce.LettuceConnection.*;
 import org.springframework.data.util.Optionals;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -277,8 +281,9 @@ public void afterPropertiesSet() {
 
 this.client = createClient();
 
-this.connectionProvider = createConnectionProvider(client, CODEC);
-this.reactiveConnectionProvider = createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC);
+this.connectionProvider = new ExceptionTranslatingConnectionProvider(createConnectionProvider(client, CODEC));
+this.reactiveConnectionProvider = new ExceptionTranslatingConnectionProvider(
+createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC));
 
 if (isClusterAware()) {
 
@@ -1222,12 +1227,7 @@ StatefulConnection<E, E> getConnection() {
  * @return the connection.
  */
 private StatefulConnection<E, E> getNativeConnection() {
-
-try {
-return connectionProvider.getConnection(StatefulConnection.class);
-} catch (RedisException e) {
-throw new RedisConnectionFailureException("Unable to connect to Redis", e);
-}
+return connectionProvider.getConnection(StatefulConnection.class);
 }
 
 /**
@@ -1418,4 +1418,124 @@ public Duration getShutdownQuietPeriod() {
 return shutdownTimeout;
 }
 }
+
+/**
+ * {@link LettuceConnectionProvider} that translates connection exceptions into {@link RedisConnectionException}.
+ */
+private static class ExceptionTranslatingConnectionProvider
+implements LettuceConnectionProvider, LettuceConnectionProvider.TargetAware, DisposableBean {
+
+private final LettuceConnectionProvider delegate;
+
+public ExceptionTranslatingConnectionProvider(LettuceConnectionProvider delegate) {
+this.delegate = delegate;
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+ */
+@Override
+public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType) {
+
+try {
+return delegate.getConnection(connectionType);
+} catch (RuntimeException e) {
+throw translateException(e);
+}
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class, RedisURI)
+ */
+@Override
+public <T extends StatefulConnection<?, ?>> T getConnection(Class<T> connectionType, RedisURI redisURI) {
+
+try {
+return ((TargetAware) delegate).getConnection(connectionType, redisURI);
+} catch (RuntimeException e) {
+throw translateException(e);
+}
+}
+
+/*
+ * (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) {
+
+CompletableFuture<T> future = new CompletableFuture<>();
+
+delegate.getConnectionAsync(connectionType).whenComplete((t, throwable) -> {
+
+if (throwable != null) {
+future.completeExceptionally(translateException(throwable));
+} else {
+future.complete(t);
+}
+});
+
+return future;
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnectionAsync(java.lang.Class, RedisURI)
+ */
+@Override
+public <T extends StatefulConnection<?, ?>> CompletionStage<T> getConnectionAsync(Class<T> connectionType,
+RedisURI redisURI) {
+
+CompletableFuture<T> future = new CompletableFuture<>();
+
+((TargetAware) delegate).getConnectionAsync(connectionType, redisURI).whenComplete((t, throwable) -> {
+
+if (throwable != null) {
+future.completeExceptionally(translateException(throwable));
+} else {
+future.complete(t);
+}
+});
+
+return future;
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection)
+ */
+@Override
+public void release(StatefulConnection<?, ?> connection) {
+delegate.release(connection);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#releaseAsync(io.lettuce.core.api.StatefulConnection)
+ */
+@Override
+public CompletableFuture<Void> releaseAsync(StatefulConnection<?, ?> connection) {
+return delegate.releaseAsync(connection);
+}
+
+/*
+ * (non-Javadoc)
+ * @see org.springframework.beans.factory.DisposableBean#destroy()
+ */
+@Override
+public void destroy() throws Exception {
+
+if (delegate instanceof DisposableBean) {
+((DisposableBean) delegate).destroy();
+}
+}
+
+private RuntimeException translateException(Throwable e) {
+return e instanceof RedisConnectionFailureException ? (RedisConnectionFailureException) e
+: new RedisConnectionFailureException("Unable to connect to Redis", e);
+}
+
+}
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

@@ -473,7 +473,8 @@ public void pubSubDoesNotSupportMasterReplicaConnections() {
 RedisConnection connection = factory.getConnection();
 
 assertThatThrownBy(() -> connection.pSubscribe((message, pattern) -> {
-}, "foo".getBytes())).isInstanceOf(RedisSystemException.class).hasCauseInstanceOf(UnsupportedOperationException.class);
+}, "foo".getBytes())).isInstanceOf(RedisConnectionFailureException.class)
+.hasCauseInstanceOf(UnsupportedOperationException.class);
 
 connection.close();
 factory.destroy();

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java

@@ -49,6 +49,8 @@
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.connection.PoolException;
 import org.springframework.data.redis.connection.RedisClusterConfiguration;
 import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConfiguration;
@@ -861,6 +863,27 @@ protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClie
 verify(connectionProviderMock, times(2)).getConnection(StatefulConnection.class);
 }
 
+@Test // DATAREDIS-1189
+public void shouldTranslateConnectionException() {
+
+LettuceConnectionProvider connectionProviderMock = mock(LettuceConnectionProvider.class);
+
+when(connectionProviderMock.getConnection(any())).thenThrow(new PoolException("error!"));
+
+LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory() {
+@Override
+protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client,
+RedisCodec<?, ?> codec) {
+return connectionProviderMock;
+}
+};
+connectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources());
+connectionFactory.afterPropertiesSet();
+
+assertThatExceptionOfType(RedisConnectionFailureException.class)
+.isThrownBy(() -> connectionFactory.getConnection().ping()).withCauseInstanceOf(PoolException.class);
+}
+
 @Test // DATAREDIS-1027
 public void shouldDisposeConnectionProviders() throws Exception {