Skip to content

[reactor-optional] Fix remaining reactive leakage #3840

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
a-TODO-rov wants to merge 1 commit into
base: feature/reactor-optional-1
Choose a base branch
from
Open

[reactor-optional] Fix remaining reactive leakage #3840

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 25 additions & 29 deletions src/main/java/io/lettuce/core/RedisClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,11 @@
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
Expand All @@ -57,7 +56,6 @@
import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import reactor.core.publisher.Mono;

/**
* A scalable and thread-safe <a href="https://redis.io/">Redis</a> client supporting synchronous, asynchronous and reactive
Expand Down Expand Up @@ -524,46 +522,44 @@ private <K, V> CompletableFuture<StatefulRedisSentinelConnection<K, V>> connectS
}

List<RedisURI> sentinels = redisURI.getSentinels();
Queue<Throwable> exceptionCollector = new LinkedBlockingQueue<>();
validateUrisAreOfSameConnectionType(sentinels);

Mono<StatefulRedisSentinelConnection<K, V>> connectionLoop = null;

for (RedisURI uri : sentinels) {

Mono<StatefulRedisSentinelConnection<K, V>> connectionMono = Mono
.fromCompletionStage(() -> doConnectSentinelAsync(codec, uri, timeout, new ConnectionMetadata(redisURI)))
.onErrorMap(CompletionException.class, Throwable::getCause)
.onErrorMap(e -> new RedisConnectionException("Cannot connect Redis Sentinel at " + uri, e))
.doOnError(exceptionCollector::add);

if (connectionLoop == null) {
connectionLoop = connectionMono;
} else {
connectionLoop = connectionLoop.onErrorResume(t -> connectionMono);
}
if (sentinels.isEmpty()) {
return Futures
.failed(new RedisConnectionException("Cannot connect to a Redis Sentinel: " + redisURI.getSentinels()));
}

if (connectionLoop == null) {
return Mono
.<StatefulRedisSentinelConnection<K, V>> error(
new RedisConnectionException("Cannot connect to a Redis Sentinel: " + redisURI.getSentinels()))
.toFuture();
List<Supplier<CompletionStage<StatefulRedisSentinelConnection<K, V>>>> attempts = new ArrayList<>(sentinels.size());
for (RedisURI uri : sentinels) {
attempts.add(() -> {
CompletableFuture<StatefulRedisSentinelConnection<K, V>> attempt = new CompletableFuture<>();
doConnectSentinelAsync(codec, uri, timeout, new ConnectionMetadata(redisURI)).whenComplete((connection, e) -> {
if (e != null) {
Throwable cause = e instanceof CompletionException && e.getCause() != null ? e.getCause() : e;
attempt.completeExceptionally(
new RedisConnectionException("Cannot connect Redis Sentinel at " + uri, cause));
} else {
attempt.complete(connection);
}
});
return attempt;
});
}

return connectionLoop.onErrorMap(e -> {
return Futures.firstSuccess(attempts, errors -> {

Throwable last = errors.get(errors.size() - 1);
RedisConnectionException ex = new RedisConnectionException(
"Cannot connect to a Redis Sentinel: " + redisURI.getSentinels(), e);
"Cannot connect to a Redis Sentinel: " + redisURI.getSentinels(), last);

for (Throwable throwable : exceptionCollector) {
if (e != throwable) {
for (Throwable throwable : errors) {
if (throwable != last) {
ex.addSuppressed(throwable);
}
}

return ex;
}).toFuture();
});
}

private <K, V> ConnectionFuture<StatefulRedisSentinelConnection<K, V>> doConnectSentinelAsync(RedisCodec<K, V> codec,
Expand Down
59 changes: 30 additions & 29 deletions src/main/java/io/lettuce/core/cluster/RedisClusterClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,6 @@
import io.lettuce.core.resource.ClientResources;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import reactor.core.publisher.Mono;

import static io.lettuce.core.RedisAuthenticationHandler.createHandler;

Expand Down Expand Up @@ -704,18 +703,14 @@ private <K, V> CompletableFuture<StatefulRedisClusterConnection<K, V>> connectCl
endpoint);
Supplier<CompletionStage<SocketAddress>> socketAddressSupplier = getSocketAddressSupplier(connection::getPartitions,
TopologyComparators::sortByClientCount);
Mono<StatefulRedisClusterConnectionImpl<K, V>> connectionMono = Mono
.defer(() -> connect(socketAddressSupplier, endpoint, connection, commandHandlerSupplier));
Supplier<CompletionStage<StatefulRedisClusterConnectionImpl<K, V>>> connectSupplier = () -> connect(
socketAddressSupplier, endpoint, connection, commandHandlerSupplier);

for (int i = 1; i < getConnectionAttempts(); i++) {
connectionMono = connectionMono
.onErrorResume(t -> connect(socketAddressSupplier, endpoint, connection, commandHandlerSupplier));
}

return connectionMono
.doOnNext(
c -> connection.registerCloseables(closeableResources, clusterWriter, pooledClusterConnectionProvider))
.map(it -> (StatefulRedisClusterConnection<K, V>) it).toFuture();
return Futures.firstSuccess(Collections.nCopies(getConnectionAttempts(), connectSupplier),

@atakavci atakavci Jul 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i acknowledge this firstSuccess looks fancy,, and function-wise it would be also pretty ok. but i suggest something more easy to follow and maintain. more intuitional..
semantics of it is a bit implicit, i mean, it is not clear whether it suggests eager execution or bind-next approach.
also in term of implementation, personally i tend to avoid recursion if it is not suggested by nature of task.
how about something like Futures.withFallback(Supplier<CompletableFuture> asycnTask, Supplier<CompletableFuture> fallbackTask).
have an overload with errorHandlers may be,, in case its needed.

errors -> errors.get(errors.size() - 1)).thenApply(it -> {
connection.registerCloseables(closeableResources, clusterWriter, pooledClusterConnectionProvider);
return (StatefulRedisClusterConnection<K, V>) it;
});
}

/**
Expand Down Expand Up @@ -756,22 +751,32 @@ protected <V, K> StatefulRedisClusterConnectionImpl<K, V> newStatefulRedisCluste
return new StatefulRedisClusterConnectionImpl(channelWriter, pushHandler, codec, timeout);
}

private <T, K, V> Mono<T> connect(Supplier<CompletionStage<SocketAddress>> socketAddressSupplier, DefaultEndpoint endpoint,
StatefulRedisClusterConnectionImpl<K, V> connection, Supplier<CommandHandler> commandHandlerSupplier) {
private <T, K, V> CompletionStage<T> connect(Supplier<CompletionStage<SocketAddress>> socketAddressSupplier,
DefaultEndpoint endpoint, StatefulRedisClusterConnectionImpl<K, V> connection,
Supplier<CommandHandler> commandHandlerSupplier) {

ConnectionFuture<T> future = connectStatefulAsync(connection, endpoint, getFirstUri(), socketAddressSupplier,
commandHandlerSupplier);

return Mono.fromCompletionStage(future).doOnError(t -> logger.warn(t.getMessage()));
return future.whenComplete((c, t) -> {
if (t != null) {
logger.warn(t.getMessage());
}
});
}

private <T, K, V> Mono<T> connect(Supplier<CompletionStage<SocketAddress>> socketAddressSupplier, DefaultEndpoint endpoint,
StatefulRedisConnectionImpl<K, V> connection, Supplier<CommandHandler> commandHandlerSupplier) {
private <T, K, V> CompletionStage<T> connect(Supplier<CompletionStage<SocketAddress>> socketAddressSupplier,
DefaultEndpoint endpoint, StatefulRedisConnectionImpl<K, V> connection,
Supplier<CommandHandler> commandHandlerSupplier) {

ConnectionFuture<T> future = connectStatefulAsync(connection, endpoint, getFirstUri(), socketAddressSupplier,
commandHandlerSupplier);

return Mono.fromCompletionStage(future).doOnError(t -> logger.warn(t.getMessage()));
return future.whenComplete((c, t) -> {
if (t != null) {
logger.warn(t.getMessage());
}
});
}

/**
Expand Down Expand Up @@ -823,18 +828,14 @@ private <K, V> CompletableFuture<StatefulRedisClusterPubSubConnection<K, V>> con
getResources(), codec, endpoint);
Supplier<CompletionStage<SocketAddress>> socketAddressSupplier = getSocketAddressSupplier(connection::getPartitions,
TopologyComparators::sortByClientCount);
Mono<StatefulRedisClusterPubSubConnectionImpl<K, V>> connectionMono = Mono
.defer(() -> connect(socketAddressSupplier, endpoint, connection, commandHandlerSupplier));
Supplier<CompletionStage<StatefulRedisClusterPubSubConnectionImpl<K, V>>> connectSupplier = () -> connect(
socketAddressSupplier, endpoint, connection, commandHandlerSupplier);

for (int i = 1; i < getConnectionAttempts(); i++) {
connectionMono = connectionMono
.onErrorResume(t -> connect(socketAddressSupplier, endpoint, connection, commandHandlerSupplier));
}

return connectionMono
.doOnNext(
c -> connection.registerCloseables(closeableResources, clusterWriter, pooledClusterConnectionProvider))
.map(it -> (StatefulRedisClusterPubSubConnection<K, V>) it).toFuture();
return Futures.firstSuccess(Collections.nCopies(getConnectionAttempts(), connectSupplier),
errors -> errors.get(errors.size() - 1)).thenApply(it -> {
connection.registerCloseables(closeableResources, clusterWriter, pooledClusterConnectionProvider);
return (StatefulRedisClusterPubSubConnection<K, V>) it;
});
}

private int getConnectionAttempts() {
Expand Down
49 changes: 49 additions & 0 deletions src/main/java/io/lettuce/core/internal/Futures.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,12 @@
package io.lettuce.core.internal;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;

import io.lettuce.core.RedisFuture;
import io.lettuce.core.resource.ClientResources;
Expand Down Expand Up @@ -303,4 +307,49 @@ public static <T> CompletableFuture<T> withTimeout(CompletableFuture<T> source,
return result;
}

/**
* Attempt a sequence of asynchronous operations in order, completing with the result of the first successful attempt. Each
* supplier is invoked lazily, only once the preceding attempt has failed. If every attempt fails, the returned future
* completes exceptionally with the result of applying {@code errorHandler} to the failures collected in attempt order.
*
* @param attempts the ordered attempts; must not be {@code null} and must not be empty.
* @param errorHandler produces the terminal failure from the collected errors; must not be {@code null}.
* @param <T> result type.
* @return a {@link CompletableFuture} completing with the first successful result or the aggregated failure.
*/
public static <T> CompletableFuture<T> firstSuccess(List<? extends Supplier<? extends CompletionStage<T>>> attempts,
Function<List<Throwable>, Throwable> errorHandler) {

CompletableFuture<T> result = new CompletableFuture<>();
attempt(attempts, 0, new ArrayList<>(), errorHandler, result);
return result;
}

private static <T> void attempt(List<? extends Supplier<? extends CompletionStage<T>>> attempts, int index,
List<Throwable> errors, Function<List<Throwable>, Throwable> errorHandler, CompletableFuture<T> result) {

if (index >= attempts.size()) {
result.completeExceptionally(errorHandler.apply(errors));
return;
}

CompletionStage<T> stage;
try {
stage = attempts.get(index).get();
} catch (Throwable t) {
errors.add(t);
attempt(attempts, index + 1, errors, errorHandler, result);
return;
}

stage.whenComplete((value, error) -> {
if (error != null) {
errors.add(error);
attempt(attempts, index + 1, errors, errorHandler, result);
} else {
result.complete(value);
}
});
}

}
65 changes: 65 additions & 0 deletions src/test/java/io/lettuce/core/internal/FuturesUnitTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,19 @@

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Supplier;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
Expand Down Expand Up @@ -165,4 +171,63 @@ void withTimeoutShouldMirrorSourceFailureWhenSourceFailsBeforeTimeout() {
assertThatThrownBy(() -> result.get(2, SECONDS)).hasCause(boom);
}

@Test
void firstSuccessReturnsFirstResultAndSkipsRemainingAttempts() throws Exception {
AtomicInteger invocations = new AtomicInteger();
List<Supplier<CompletionStage<String>>> attempts = Arrays.asList(() -> {
invocations.incrementAndGet();
return CompletableFuture.completedFuture("first");
}, () -> {
invocations.incrementAndGet();
return CompletableFuture.completedFuture("second");
});

CompletableFuture<String> result = Futures.firstSuccess(attempts, errors -> new IllegalStateException());

assertThat(result.get(2, SECONDS)).isEqualTo("first");
assertThat(invocations).hasValue(1);
}

@Test
void firstSuccessFallsThroughToLaterAttempt() throws Exception {
List<Supplier<CompletionStage<String>>> attempts = Arrays.asList(() -> Futures.failed(new RuntimeException("nope")),
() -> CompletableFuture.completedFuture("recovered"));

CompletableFuture<String> result = Futures.firstSuccess(attempts, errors -> new IllegalStateException());

assertThat(result.get(2, SECONDS)).isEqualTo("recovered");
}

@Test
void firstSuccessAggregatesFailuresInOrderWhenAllFail() {
RuntimeException e1 = new RuntimeException("e1");
RuntimeException e2 = new RuntimeException("e2");
List<Supplier<CompletionStage<String>>> attempts = Arrays.asList(() -> Futures.failed(e1), () -> Futures.failed(e2));

Function<List<Throwable>, Throwable> aggregator = errors -> {
Throwable last = errors.get(errors.size() - 1);
RuntimeException aggregate = new RuntimeException("all failed", last);
for (Throwable t : errors) {
if (t != last) {
aggregate.addSuppressed(t);
}
}
return aggregate;
};

CompletableFuture<String> result = Futures.firstSuccess(attempts, aggregator);

Throwable aggregate = null;
try {
result.get(2, SECONDS);
} catch (ExecutionException e) {
aggregate = e.getCause();
} catch (InterruptedException | TimeoutException e) {
Thread.currentThread().interrupt();
}

assertThat(aggregate).hasMessage("all failed").hasCause(e2);
assertThat(aggregate.getSuppressed()).containsExactly(e1);
}

}
Loading