Use Mono.usingWhen(…) to avoid connection leaks on master lookup thro…

…ugh Sentinel #1558 Previously, we used our own hooks to open and close the lookup connection that could leave connections open when no master node was found.
redis · Jan 14, 2021 · c3337f5 · c3337f5
1 parent 1de2cf7
commit c3337f5
Showing 1 changed file with 8 additions and 9 deletions.
diff --git a/src/main/java/io/lettuce/core/RedisClient.java b/src/main/java/io/lettuce/core/RedisClient.java
@@ -740,10 +740,8 @@ private Mono<SocketAddress> lookupRedis(RedisURI sentinelUri) {
 
         Duration timeout = getDefaultTimeout();
 
-        Mono<StatefulRedisSentinelConnection<String, String>> connection = Mono
-                .fromCompletionStage(() -> connectSentinelAsync(newStringStringCodec(), sentinelUri, timeout));
-
-        return connection.flatMap(c -> {
+        return Mono.usingWhen(
+                Mono.fromCompletionStage(() -> connectSentinelAsync(newStringStringCodec(), sentinelUri, timeout)), c -> {
 
             String sentinelMasterId = sentinelUri.getSentinelMasterId();
             return c.reactive().getMasterAddrByName(sentinelMasterId).map(it -> {
@@ -762,16 +760,17 @@ private Mono<SocketAddress> lookupRedis(RedisURI sentinelUri) {
 
                 return it;
             }).timeout(timeout) //
-                    .onErrorResume(e -> {
+                            .onErrorMap(e -> {
 
                         RedisCommandTimeoutException ex = ExceptionFactory
                                 .createTimeoutException("Cannot obtain master using SENTINEL MASTER", timeout);
                         ex.addSuppressed(e);
 
-                        return Mono.fromCompletionStage(c::closeAsync).then(Mono.error(ex));
-                    }).flatMap(it -> Mono.fromCompletionStage(c::closeAsync) //
-                            .thenReturn(it));
-        });
+                                return ex;
+                            });
+                }, c -> Mono.fromCompletionStage(c::closeAsync), //
+                (c, ex) -> Mono.fromCompletionStage(c::closeAsync), //
+                c -> Mono.fromCompletionStage(c::closeAsync));
     }
 
     private static <T> ConnectionFuture<T> transformAsyncConnectionException(ConnectionFuture<T> future) {