[reactor-optional] Drop 'Flux' (Reactor) dependency from RedisCredentialsProvider#3785
Conversation
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 3 potential issues.
Reviewed by Cursor Bugbot for commit de2287f. Configure here.
| CompletableFuture<RedisCredentials> previous = credentialsFutureRef.get(); | ||
| if (!previous.isDone()) { | ||
| credentialsFutureRef.compareAndSet(previous, new CompletableFuture<>()); | ||
| executor.execute(() -> previous.completeExceptionally(exception)); |
There was a problem hiding this comment.
Failed CAS still fails future
Medium Severity
In onError, compareAndSet on credentialsFutureRef is not checked, yet previous is always completed exceptionally on the executor. If renewal already swapped the ref, a stale pending future can be failed after a successful token update, so resolveCredentials waiters may see an error while the provider already holds fresh credentials.
Reviewed by Cursor Bugbot for commit de2287f. Configure here.
There was a problem hiding this comment.
@atakavci sounds like a problem, but not sure ...
| @Override | ||
| public void close() { | ||
| provider.subscriptions.remove(this); | ||
| } |
There was a problem hiding this comment.
Closed subscriptions still get callbacks
Medium Severity
SimpleSubscription.close() only removes the subscription from the list; it does not invalidate the handle. Tasks already submitted to the executor can still run dispatchOnNext or dispatchOnError, so RedisAuthenticationHandler may re-authenticate or report errors after unsubscribe() or provider close().
Additional Locations (1)
Reviewed by Cursor Bugbot for commit de2287f. Configure here.
There was a problem hiding this comment.
@atakavci don't you think we need a volatile isClosed flag here ?
There was a problem hiding this comment.
| SimpleSubscription subscription = new SimpleSubscription(this, onNext, onError); | ||
| subscriptions.add(subscription); | ||
| executor.execute(() -> subscription.replay(getReplayCandidate())); | ||
| return subscription; |
There was a problem hiding this comment.
Subscribe after close race
Low Severity
subscribeToCredentials checks isClosed once at entry, but close() can run before the subscription is added or before replay is scheduled. A subscription can remain registered after the provider is closed and still receive a replay delivery.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit de2287f. Configure here.
There was a problem hiding this comment.
Pull request overview
This PR removes Project Reactor types from the public RedisCredentialsProvider API by switching credential resolution to CompletionStage and replacing the streaming Flux API with a callback-based subscription (subscribeToCredentials(…, …) returning a CredentialsSubscription). It updates core authentication/handshake code paths, the token-based credentials provider implementation, and the corresponding tests and documentation to match the new non-Reactor surface.
Changes:
- Reworks
RedisCredentialsProviderto useCompletionStageforresolveCredentials()and a consumer-based streaming API withCredentialsSubscription. - Updates authentication/handshake consumers (
RedisAuthenticationHandler,RedisHandshake,RedisURI) and supporting providers (StaticCredentialsProvider,TokenBasedRedisCredentialsProvider) to the new model. - Migrates affected tests and the user guide examples away from
Mono/Flux-based credential resolution and streaming.
Reviewed changes
Copilot reviewed 19 out of 19 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| src/test/java/io/lettuce/core/sentinel/SentinelAclIntegrationTests.java | Updates sentinel ACL tests to use CompletableFuture-based credentials provider. |
| src/test/java/io/lettuce/core/RedisURIUnitTests.java | Adds/updates tests for RedisURI credential-provider error unwrapping and switches to CompletionStage. |
| src/test/java/io/lettuce/core/RedisURIBuilderUnitTests.java | Adjusts builder tests to verify credentials via Mono.fromCompletionStage(...). |
| src/test/java/io/lettuce/core/RedisHandshakeUnitTests.java | Updates test credentials provider to return CompletionStage rather than Mono. |
| src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java | Replaces Reactor-based streaming example provider with callback subscription + CompletableFuture latest-value storage. |
| src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java | Updates integration test to resolve credentials via CompletionStage instead of Mono.block(). |
| src/test/java/io/lettuce/core/cluster/RedisClusterURIUtilUnitTests.java | Migrates cluster URI tests to CompletionStage credential resolution. |
| src/test/java/io/lettuce/core/AuthenticationIntegrationTests.java | Updates custom credentials provider usage to CompletableFuture.completedFuture(...). |
| src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java | Rewrites streaming tests to use the new subscription API and CompletionStage resolution. |
| src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java | Updates Entra ID renewal tests to use CredentialsSubscription and ensures cleanup via close(). |
| src/test/java/io/lettuce/authx/DefaultAzureCredentialsIntegrationTests.java | Switches token credential resolution to CompletableFuture#get with timeout. |
| src/main/java/io/lettuce/core/StaticCredentialsProvider.java | Replaces cached Mono with a cached completed CompletableFuture. |
| src/main/java/io/lettuce/core/RedisURI.java | Removes Reactor usage; masks credentials by synchronously joining the CompletionStage and bubbling exceptions. |
| src/main/java/io/lettuce/core/RedisHandshake.java | Replaces Mono.toFuture() with CompletionStage.toCompletableFuture() for handshake auth. |
| src/main/java/io/lettuce/core/RedisCredentialsProvider.java | Core public API change: CompletionStage + callback subscription model + new CredentialsSubscription. |
| src/main/java/io/lettuce/core/RedisAuthenticationHandler.java | Replaces Reactor Disposable subscription management with CredentialsSubscription lifecycle. |
| src/main/java/io/lettuce/core/ClientOptions.java | Updates reauthentication javadoc to reference subscribeToCredentials(...). |
| src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java | Reimplements streaming token-based provider without Reactor; adds executor-based dispatch and subscription replay behavior. |
| docs/user-guide/connecting-redis.md | Updates streaming credentials documentation example to the new subscription API and CompletionStage resolution. |
Comments suppressed due to low confidence (1)
src/test/java/io/lettuce/core/AuthenticationIntegrationTests.java:37
- The new
CompletableFutureimport breaks the localjava.*import ordering (it’s inserted betweenjava.time.*imports andjava.util.*). Re-sorting keeps imports consistent and avoids style-check failures.
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| return () -> { | ||
| try { | ||
| return CompletableFuture.completedFuture(supplier.get()); | ||
| } catch (Exception e) { | ||
| return Futures.failed(e); | ||
| } | ||
| }; |
| default CompletionStage<RedisCredentials> resolveCredentials() { | ||
| try { | ||
| return CompletableFuture.completedFuture(resolveCredentialsNow()); | ||
| } catch (Exception e) { | ||
| return Futures.failed(e); | ||
| } | ||
| } |
There was a problem hiding this comment.
@atakavci maybe we should consider null as failed Future ?
| public void unsubscribe() { | ||
| Disposable subscription = credentialsSubscription.getAndSet(null); | ||
| if (subscription != null && !subscription.isDisposed()) { | ||
| subscription.dispose(); | ||
| CredentialsSubscription sub = credentialsSubscription.getAndSet(null); | ||
| if (sub != null) { | ||
| sub.close(); | ||
| } |
| @Override | ||
| public CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) { | ||
| Listener listener = new Listener(onNext, onError); |
|
|
||
| import java.util.concurrent.CompletableFuture; | ||
|
|
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| provider.subscriptions.remove(this); | ||
| } | ||
|
|
||
| private void onCredentials(RedisCredentials credentials) { | ||
| RedisCredentials played = lastPlayed.getAndSet(credentials); | ||
| if (played != credentials) { | ||
| onNext.accept(credentials); | ||
| } | ||
| } | ||
|
|
||
| private void replay(RedisCredentials candidate) { | ||
| if (candidate != null && lastPlayed.compareAndSet(null, candidate)) { | ||
| onNext.accept(candidate); | ||
| } | ||
| } | ||
|
|
||
| private void onError(Throwable throwable) { | ||
| onError.accept(throwable); | ||
| } |
There was a problem hiding this comment.
| /** | ||
| * Handle to a subscription created by {@link #subscribeToCredentials(Consumer, Consumer)}. Closing the subscription stops | ||
| * the provider from delivering further credential updates to the registered consumers. | ||
| */ | ||
| interface CredentialsSubscription extends Closeable { | ||
|
|
||
| @Override | ||
| void close(); | ||
|
|
||
| } |
There was a problem hiding this comment.
I suspect his will be reused across Flux replacements. For example - https://github.com/redis/lettuce/pull/3789/changes#diff-6b86d8999bb040fc318c29072b8fa3488fe38f7959bd8d400d5bf0a04775b677
| @Override | ||
| public void close() { | ||
| provider.subscriptions.remove(this); | ||
| } |
There was a problem hiding this comment.
@atakavci don't you think we need a volatile isClosed flag here ?
| @Override | ||
| public void close() { | ||
| provider.subscriptions.remove(this); | ||
| } |
There was a problem hiding this comment.
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| provider.subscriptions.remove(this); | ||
| } | ||
|
|
||
| private void onCredentials(RedisCredentials credentials) { | ||
| RedisCredentials played = lastPlayed.getAndSet(credentials); | ||
| if (played != credentials) { | ||
| onNext.accept(credentials); | ||
| } | ||
| } | ||
|
|
||
| private void replay(RedisCredentials candidate) { | ||
| if (candidate != null && lastPlayed.compareAndSet(null, candidate)) { | ||
| onNext.accept(candidate); | ||
| } | ||
| } | ||
|
|
||
| private void onError(Throwable throwable) { | ||
| onError.accept(throwable); | ||
| } |
There was a problem hiding this comment.
| CompletableFuture<RedisCredentials> previous = credentialsFutureRef.get(); | ||
| if (!previous.isDone()) { | ||
| credentialsFutureRef.compareAndSet(previous, new CompletableFuture<>()); | ||
| executor.execute(() -> previous.completeExceptionally(exception)); |
There was a problem hiding this comment.
@atakavci sounds like a problem, but not sure ...
| default CompletionStage<RedisCredentials> resolveCredentials() { | ||
| try { | ||
| return CompletableFuture.completedFuture(resolveCredentialsNow()); | ||
| } catch (Exception e) { | ||
| return Futures.failed(e); | ||
| } | ||
| } |
There was a problem hiding this comment.
@atakavci maybe we should consider null as failed Future ?
Removes Project Reactor (
Flux/Sinks/Disposable) from the streaming credentials API exposed byRedisCredentialsProvider, replacing it with a callback-based subscription model built onCompletableFutureandConsumer. The credentials-provider surface no longer referencesreactor-coretypes.API changes
io.lettuce.core.RedisCredentialsProvider:default Flux<RedisCredentials> credentials().default CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError), still throwingUnsupportedOperationExceptionwhensupportsStreaming()isfalse.CredentialsSubscription extends Closeable(non-throwingclose()) as the subscription handle.io.lettuce.core.RedisAuthenticationHandler: subscription state moves fromAtomicReference<Disposable>toAtomicReference<CredentialsSubscription>;subscribe()callssubscribeToCredentials(this::reauthenticate, this::onError)and closes any previously held subscription;unsubscribe()closes it. InternalonNext/completehelpers are dropped.io.lettuce.core.ClientOptions:ReauthenticateBehaviorjavadoc updated fromcredentials()tosubscribeToCredentials(...).Implementation
io.lettuce.authx.TokenBasedRedisCredentialsProviderreplacesSinks.Many<RedisCredentials>with:AtomicReference<CompletableFuture<RedisCredentials>>for the latest credentials (backsresolveCredentials()and subscriber replay).CopyOnWriteArrayList<SimpleSubscription>for live subscribers; nestedSimpleSubscriptiondeduplicates replay vs. live delivery and self-removes onclose().Executorfor dispatchingonNext/onErrorand completing the initial future.TokenListener.onTokenRenewedupdates the latest future and fans out to current subscriptions via the executor.TokenListener.onErrorlogs, fails the pending future once (resetting the ref so later calls wait for the next success), and dispatches to current subscribers; prior errors are not replayed to later subscribers.close()setsisClosed, stops theTokenManager, fails any pending future withIllegalStateException("Credentials provider closed")and clears subscriptions; listener callbacks short-circuit when closed.resolveCredentials()returns a freshCompletableFuturechained off the current one.New factory overloads
create(TokenAuthConfig, Executor)andcreate(TokenManager, Executor); the no-executor variants default to a caller-thread executor (r -> r.run()). Javadoc spells out the threading consequences and the per-subscription ordering requirement.Documentation
docs/user-guide/connecting-redis.md: streaming-credentials example rewritten on top ofsubscribeToCredentials(...)with aCompletableFuture-backed latest-credentials reference and a plain listener list.Backwards compatibility
Breaking change for any external implementer or caller of
RedisCredentialsProvider#credentials(). Implementers should overridesubscribeToCredentials(Consumer, Consumer)(andsupportsStreaming()); callers should subscribe via the consumer API and close the returnedCredentialsSubscription.Note
High Risk
This is a breaking public API change in authentication and token-renewal paths, with new threading semantics in
TokenBasedRedisCredentialsProviderthat can affect re-auth timing if subscribers block or use unordered executors.Overview
Breaking API change:
RedisCredentialsProviderno longer exposes Reactor types.resolveCredentials()returnsCompletionStage<RedisCredentials>instead ofMono, and streaming updates move from removedcredentials()tosubscribeToCredentials(onNext, onError)returning a closableCredentialsSubscription.Driver wiring:
RedisAuthenticationHandlersubscribes via the new callback API (replacingFlux+Disposable).ReauthenticateBehaviordocs now referencesubscribeToCredentials. Handshake,RedisURI, and static providers useCompletableFuture/toCompletableFuture()instead ofMono/block().Token / Entra provider:
TokenBasedRedisCredentialsProviderdropsSinksfor an atomic latest-credentials future, subscriber list, and optionalExecutor-dispatched callbacks; newcreate(..., Executor)overloads document threading and ordering trade-offs.Docs & tests: User guide sample rewritten on the callback model; tests and integration code updated accordingly. External callers/implementers of
credentials()must migrate to the subscription API.Reviewed by Cursor Bugbot for commit 2a8afaf. Bugbot is set up for automated code reviews on this repo. Configure here.