Skip to content

Fix ClientOptions race condition during async connection #3891

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
aartisonigra wants to merge 1 commit into
base: main
Choose a base branch
from
+166 −77
Open

Fix ClientOptions race condition during async connection #3891

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
4 changes: 4 additions & 0 deletions src/main/java/io/lettuce/core/AbstractRedisClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -597,6 +597,10 @@ private CompletableFuture<Void> closeClientResources(long quietPeriod, long time
}

protected RedisHandshake createHandshake(ConnectionState state) {
return createHandshake(state, clientOptions);
}

protected RedisHandshake createHandshake(ConnectionState state, ClientOptions clientOptions) {
EndpointTypeSource source = null;
if (clientOptions.getMaintNotificationsConfig().maintNotificationsEnabled()) {
LettuceAssert.notNull(clientOptions.getMaintNotificationsConfig().getEndpointTypeSource(),
Expand Down
239 changes: 162 additions & 77 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 @@ -97,6 +97,8 @@ public class RedisClient extends AbstractRedisClient {

private static final RedisURI EMPTY_URI = new RedisURI();

private final ThreadLocal<ClientOptions> clientOptionsThreadLocal = new ThreadLocal<>();

private final RedisURI redisURI;

protected RedisClient(ClientResources clientResources, RedisURI redisURI) {
Expand Down Expand Up @@ -276,36 +278,42 @@ private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandalone

logger.debug("Trying to get a Redis connection for: {}", redisURI);

DefaultEndpoint endpoint = createEndpoint();
RedisChannelWriter writer = endpoint;
ClientOptions clientOptions = getOptions();
clientOptionsThreadLocal.set(clientOptions);
try {
DefaultEndpoint endpoint = createEndpoint();

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Pass the captured options into endpoint creation

Although this method captures ClientOptions before entering the setup block, the endpoint is still built through createEndpoint(), whose default implementation calls getOptions() again. If setOptions() runs in that window, the DefaultEndpoint can pick up newer request-queue/reconnect settings while the writer, handler, auth, and handshake use the captured options, so a single new connection still observes mixed options; pass the captured options into the endpoint factory, and do the same for the Pub/Sub endpoint factory, to close the race.

Useful? React with 👍 / 👎.

RedisChannelWriter writer = endpoint;

if (CommandExpiryWriter.isSupported(getOptions())) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
}
if (CommandExpiryWriter.isSupported(clientOptions)) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, clientOptions, getResources());
}

if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}
if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}

StatefulRedisConnectionImpl<K, V> connection = newStatefulRedisConnection(writer, endpoint, codec, timeout);
StatefulRedisConnectionImpl<K, V> connection = newStatefulRedisConnection(writer, endpoint, codec, timeout,
clientOptions);
Comment on lines +295 to +296

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P1 Badge Preserve regular connection factory overrides

For clients that subclass RedisClient and override the existing 4-argument newStatefulRedisConnection(...) hook, this changed call now statically selects the new 5-argument overload on the base class, so connect()/connectAsync() silently stop creating the subclass connection. Bridge existing overrides when adding the ClientOptions-aware overload so normal connection customization remains effective.

Useful? React with 👍 / 👎.


ClientOptions clientOptions = getOptions();
ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(connection, endpoint, redisURI,
() -> new CommandHandler(clientOptions, getResources(), endpoint), false);
ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(connection, endpoint, redisURI,
() -> new CommandHandler(clientOptions, getResources(), endpoint), false, clientOptions);

future.whenComplete((channelHandler, throwable) -> {
future.whenComplete((channelHandler, throwable) -> {

if (throwable != null) {
connection.closeAsync();
}
});
if (throwable != null) {
connection.closeAsync();
}
});

return future;
return future;
} finally {
clientOptionsThreadLocal.remove();
}
}

@SuppressWarnings("unchecked")
private <K, V, S> ConnectionFuture<S> connectStatefulAsync(StatefulRedisConnectionImpl<K, V> connection, Endpoint endpoint,
RedisURI redisURI, Supplier<CommandHandler> commandHandlerSupplier, Boolean isPubSub) {
RedisURI redisURI, Supplier<CommandHandler> commandHandlerSupplier, Boolean isPubSub, ClientOptions clientOptions) {

ConnectionBuilder connectionBuilder;
if (redisURI.isSsl()) {
Expand All @@ -319,15 +327,15 @@ private <K, V, S> ConnectionFuture<S> connectStatefulAsync(StatefulRedisConnecti
ConnectionState state = connection.getConnectionState();
state.apply(redisURI);
state.setDb(redisURI.getDatabase());
connection
.setAuthenticationHandler(createHandler(connection, redisURI.getCredentialsProvider(), isPubSub, getOptions()));
connection.setAuthenticationHandler(
createHandler(connection, redisURI.getCredentialsProvider(), isPubSub, clientOptions));
connectionBuilder.connection(connection);
connectionBuilder.clientOptions(getOptions());
connectionBuilder.clientOptions(clientOptions);
connectionBuilder.clientResources(getResources());
connectionBuilder.commandHandler(commandHandlerSupplier).endpoint(endpoint);

connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, connection.getConnectionEvents(), redisURI);
connectionBuilder.connectionInitializer(createHandshake(state));
connectionBuilder.connectionInitializer(createHandshake(state, clientOptions));

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Keep invoking overridden handshake factories

If a subclass overrides the existing protected createHandshake(ConnectionState) hook to customize activation, this direct call to the new two-argument overload bypasses that override for standalone connections. In those subclassed clients the custom handshake is ignored even though the code still compiles, so adapt the captured-options path without skipping existing one-argument overrides.

Useful? React with 👍 / 👎.


ConnectionFuture<RedisChannelHandler<K, V>> future = initializeChannelAsync(connectionBuilder);

Expand Down Expand Up @@ -410,29 +418,35 @@ private <K, V> ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectPubS
assertNotNull(codec);
checkValidRedisURI(redisURI);

PubSubEndpoint<K, V> endpoint = createPubSubEndpoint();
RedisChannelWriter writer = endpoint;
ClientOptions clientOptions = getOptions();
clientOptionsThreadLocal.set(clientOptions);
try {
PubSubEndpoint<K, V> endpoint = createPubSubEndpoint();
RedisChannelWriter writer = endpoint;

if (CommandExpiryWriter.isSupported(getOptions())) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
}
if (CommandExpiryWriter.isSupported(clientOptions)) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, clientOptions, getResources());
}

if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}
if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}

StatefulRedisPubSubConnectionImpl<K, V> connection = newStatefulRedisPubSubConnection(endpoint, writer, codec, timeout);
StatefulRedisPubSubConnectionImpl<K, V> connection = newStatefulRedisPubSubConnection(endpoint, writer, codec,
timeout, clientOptions);

@cursor cursor Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Subclass factory hooks bypassed

High Severity

Async connect paths now call the new five-argument newStatefulRedis* factories directly, so subclasses that override the documented four-argument hooks are never invoked. That breaks existing extensibility, including MyExtendedRedisClient and its integration test that expects a MyPubSubConnection.

Additional Locations (2)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 49e24f1. Configure here.

Comment on lines +435 to +436

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P1 Badge Preserve existing Pub/Sub connection factory overrides

With subclasses that override only the existing 4-argument newStatefulRedisPubSubConnection(...) hook, such as MyExtendedRedisClient in the extensibility test, this call now dispatches to the new 5-argument overload instead, whose default returns StatefulRedisPubSubConnectionImpl. That bypasses the documented extension point and makes connectPubSub() stop returning the subclass connection, so the existing MyPubSubConnection assertion path regresses unless the new flow preserves or delegates to the old hook.

Useful? React with 👍 / 👎.


ClientOptions clientOptions = getOptions();
ConnectionFuture<StatefulRedisPubSubConnection<K, V>> future = connectStatefulAsync(connection, endpoint, redisURI,
() -> new PubSubCommandHandler<>(clientOptions, getResources(), codec, endpoint), true);
ConnectionFuture<StatefulRedisPubSubConnection<K, V>> future = connectStatefulAsync(connection, endpoint, redisURI,
() -> new PubSubCommandHandler<>(clientOptions, getResources(), codec, endpoint), true, clientOptions);

return future.whenComplete((conn, throwable) -> {
return future.whenComplete((conn, throwable) -> {

if (throwable != null) {
conn.close();
}
});
if (throwable != null) {
conn.close();
}
});
} finally {
clientOptionsThreadLocal.remove();
}
}

/**
Expand Down Expand Up @@ -567,52 +581,59 @@ private <K, V> CompletableFuture<StatefulRedisSentinelConnection<K, V>> connectS
private <K, V> ConnectionFuture<StatefulRedisSentinelConnection<K, V>> doConnectSentinelAsync(RedisCodec<K, V> codec,
RedisURI redisURI, Duration timeout, ConnectionMetadata metadata) {

ConnectionBuilder connectionBuilder;
if (redisURI.isSsl()) {
SslConnectionBuilder sslConnectionBuilder = SslConnectionBuilder.sslConnectionBuilder();
sslConnectionBuilder.ssl(redisURI);
connectionBuilder = sslConnectionBuilder;
} else {
connectionBuilder = ConnectionBuilder.connectionBuilder();
}
connectionBuilder.clientOptions(ClientOptions.copyOf(getOptions()));
connectionBuilder.clientResources(getResources());
ClientOptions clientOptions = getOptions();
clientOptionsThreadLocal.set(clientOptions);
try {
ConnectionBuilder connectionBuilder;
if (redisURI.isSsl()) {
SslConnectionBuilder sslConnectionBuilder = SslConnectionBuilder.sslConnectionBuilder();
sslConnectionBuilder.ssl(redisURI);
connectionBuilder = sslConnectionBuilder;
} else {
connectionBuilder = ConnectionBuilder.connectionBuilder();
}
connectionBuilder.clientOptions(ClientOptions.copyOf(clientOptions));
connectionBuilder.clientResources(getResources());

DefaultEndpoint endpoint = createEndpoint();
RedisChannelWriter writer = endpoint;
DefaultEndpoint endpoint = createEndpoint();
RedisChannelWriter writer = endpoint;

if (CommandExpiryWriter.isSupported(getOptions())) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
}
if (CommandExpiryWriter.isSupported(clientOptions)) {
writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, clientOptions, getResources());
}

if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}
if (CommandListenerWriter.isSupported(getCommandListeners())) {
writer = new CommandListenerWriter(writer, getCommandListeners());
}

StatefulRedisSentinelConnectionImpl<K, V> connection = newStatefulRedisSentinelConnection(writer, codec, timeout);
ConnectionState state = connection.getConnectionState();
StatefulRedisSentinelConnectionImpl<K, V> connection = newStatefulRedisSentinelConnection(writer, codec, timeout,
clientOptions);
Comment on lines +609 to +610

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P1 Badge Preserve Sentinel connection factory overrides

For subclasses that override the existing 3-argument newStatefulRedisSentinelConnection(...) hook, this call now binds to the new 4-argument overload on RedisClient instead of dispatching through the overridden method. Sentinel connections from those clients therefore revert to the default StatefulRedisSentinelConnectionImpl, so the new overload should be bridged without bypassing existing custom factories.

Useful? React with 👍 / 👎.

ConnectionState state = connection.getConnectionState();

state.apply(redisURI);
state.apply(metadata);
state.apply(redisURI);
state.apply(metadata);

connectionBuilder.connectionInitializer(createHandshake(state));
connectionBuilder.connectionInitializer(createHandshake(state, clientOptions));

logger.debug("Connecting to Redis Sentinel, address: " + redisURI);
logger.debug("Connecting to Redis Sentinel, address: " + redisURI);

ClientOptions clientOptions = getOptions();
connectionBuilder.endpoint(endpoint).commandHandler(() -> new CommandHandler(clientOptions, getResources(), endpoint))
.connection(connection);
connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, connection.getConnectionEvents(), redisURI);
connectionBuilder.endpoint(endpoint)
.commandHandler(() -> new CommandHandler(clientOptions, getResources(), endpoint)).connection(connection);
connectionBuilder(getSocketAddressSupplier(redisURI), connectionBuilder, connection.getConnectionEvents(),
redisURI);

ConnectionFuture<?> sync = initializeChannelAsync(connectionBuilder);
ConnectionFuture<?> sync = initializeChannelAsync(connectionBuilder);

return sync.thenApply(ignore -> (StatefulRedisSentinelConnection<K, V>) connection).whenComplete((ignore, e) -> {
return sync.thenApply(ignore -> (StatefulRedisSentinelConnection<K, V>) connection).whenComplete((ignore, e) -> {

if (e != null) {
logger.warn("Cannot connect Redis Sentinel at " + redisURI + ": " + e);
connection.closeAsync();
}
});
if (e != null) {
logger.warn("Cannot connect Redis Sentinel at " + redisURI + ": " + e);
connection.closeAsync();
}
});
} finally {
clientOptionsThreadLocal.remove();
}
}

/**
Expand Down Expand Up @@ -645,6 +666,25 @@ public void setOptions(ClientOptions clientOptions) {
*/
protected <K, V> StatefulRedisPubSubConnectionImpl<K, V> newStatefulRedisPubSubConnection(PubSubEndpoint<K, V> endpoint,
RedisChannelWriter channelWriter, RedisCodec<K, V> codec, Duration timeout) {
return newStatefulRedisPubSubConnection(endpoint, channelWriter, codec, timeout, getOptions());
}

/**
* Create a new instance of {@link StatefulRedisPubSubConnectionImpl} or a subclass.
* <p>
* Subclasses of {@link RedisClient} may override that method.
*
* @param endpoint the endpoint
* @param channelWriter the channel writer
* @param codec codec
* @param timeout default timeout
* @param clientOptions the client options
* @param <K> Key-Type
* @param <V> Value Type
* @return new instance of StatefulRedisPubSubConnectionImpl
*/
protected <K, V> StatefulRedisPubSubConnectionImpl<K, V> newStatefulRedisPubSubConnection(PubSubEndpoint<K, V> endpoint,
RedisChannelWriter channelWriter, RedisCodec<K, V> codec, Duration timeout, ClientOptions clientOptions) {

@cursor cursor Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

PubSub options parameter unused

Medium Severity

The new newStatefulRedisPubSubConnection overload accepts clientOptions but never uses it, and still builds the connection with the default JSON parser. The four-argument overload also re-reads getOptions() instead of the ThreadLocal used by the standalone and sentinel factories, so Pub/Sub stays inconsistent with the rest of this fix.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 49e24f1. Configure here.

Comment on lines +686 to +687

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Add @SInCE to the new protected overloads

This newly added protected overload is an extension point in RedisClient, but its Javadoc omits @since; the other new ClientOptions-aware factory overloads do the same. Add the current release tag (7.7 after dropping -SNAPSHOT) so generated API docs remain versioned for new public API.

AGENTS.md reference: AGENTS.md:L145-L146

Useful? React with 👍 / 👎.

return new StatefulRedisPubSubConnectionImpl<>(endpoint, channelWriter, codec, timeout);
}

Expand All @@ -662,7 +702,29 @@ protected <K, V> StatefulRedisPubSubConnectionImpl<K, V> newStatefulRedisPubSubC
*/
protected <K, V> StatefulRedisSentinelConnectionImpl<K, V> newStatefulRedisSentinelConnection(
RedisChannelWriter channelWriter, RedisCodec<K, V> codec, Duration timeout) {
return new StatefulRedisSentinelConnectionImpl<>(channelWriter, codec, timeout, getOptions().getJsonParser());
ClientOptions clientOptions = clientOptionsThreadLocal.get();
if (clientOptions == null) {
clientOptions = getOptions();
}
return newStatefulRedisSentinelConnection(channelWriter, codec, timeout, clientOptions);
}

/**
* Create a new instance of {@link StatefulRedisSentinelConnectionImpl} or a subclass.
* <p>
* Subclasses of {@link RedisClient} may override that method.
*
* @param channelWriter the channel writer
* @param codec codec
* @param timeout default timeout
* @param clientOptions the client options
* @param <K> Key-Type
* @param <V> Value Type
* @return new instance of StatefulRedisSentinelConnectionImpl
*/
protected <K, V> StatefulRedisSentinelConnectionImpl<K, V> newStatefulRedisSentinelConnection(
RedisChannelWriter channelWriter, RedisCodec<K, V> codec, Duration timeout, ClientOptions clientOptions) {
return new StatefulRedisSentinelConnectionImpl<>(channelWriter, codec, timeout, clientOptions.getJsonParser());
}

/**
Expand All @@ -680,7 +742,30 @@ protected <K, V> StatefulRedisSentinelConnectionImpl<K, V> newStatefulRedisSenti
*/
protected <K, V> StatefulRedisConnectionImpl<K, V> newStatefulRedisConnection(RedisChannelWriter channelWriter,
PushHandler pushHandler, RedisCodec<K, V> codec, Duration timeout) {
return new StatefulRedisConnectionImpl<>(channelWriter, pushHandler, codec, timeout, getOptions().getJsonParser());
ClientOptions clientOptions = clientOptionsThreadLocal.get();
if (clientOptions == null) {
clientOptions = getOptions();
}
return newStatefulRedisConnection(channelWriter, pushHandler, codec, timeout, clientOptions);
}

/**
* Create a new instance of {@link StatefulRedisConnectionImpl} or a subclass.
* <p>
* Subclasses of {@link RedisClient} may override that method.
*
* @param channelWriter the channel writer
* @param pushHandler the handler for push notifications
* @param codec codec
* @param timeout default timeout
* @param clientOptions the client options
* @param <K> Key-Type
* @param <V> Value Type
* @return new instance of StatefulRedisConnectionImpl
*/
protected <K, V> StatefulRedisConnectionImpl<K, V> newStatefulRedisConnection(RedisChannelWriter channelWriter,
PushHandler pushHandler, RedisCodec<K, V> codec, Duration timeout, ClientOptions clientOptions) {
return new StatefulRedisConnectionImpl<>(channelWriter, pushHandler, codec, timeout, clientOptions.getJsonParser());

@cursor cursor Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Endpoint still races on options

High Severity

createEndpoint and createPubSubEndpoint still call getOptions() after options were already snapshotted for the connection. Concurrent setOptions() can leave DefaultEndpoint/PubSubEndpoint with different ClientOptions than CommandHandler, handshake, and ConnectionBuilder, which is the inconsistency issue #3591 calls out.

Additional Locations (2)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 49e24f1. Configure here.

}

/**
Expand Down
Loading