[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.
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.
| /** | ||
| * 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
There was a problem hiding this comment.
i will leave this to you , to replace or move to another place.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
Reviewed by Cursor Bugbot for commit 18e0954. Configure here.
| if (!previous.isDone()) { | ||
| executor.execute(() -> previous.complete(credentials)); | ||
| } | ||
| subscriptions.forEach(s -> executor.execute(() -> dispatchOnNext(s, credentials))); |
There was a problem hiding this comment.
Renewal after close revives credentials
Medium Severity
In TokenBasedRedisCredentialsProvider, onTokenRenewed only checks isClosed once at entry, then always swaps in a successfully completed credentialsFutureRef. An in-flight renewal that passes that check can run after close() has set isClosed, stopped the manager, and failed the pending future—restoring valid credentials and letting resolveCredentials() succeed on a closed provider.
Reviewed by Cursor Bugbot for commit 18e0954. Configure here.
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
Breaking public API for custom credential providers and any code using
credentials()orMonoresolution; changes affect connection handshake and token-based Entra ID re-auth paths.Overview
Breaking API change:
RedisCredentialsProviderno longer exposes Reactor types.resolveCredentials()returnsCompletionStage<RedisCredentials>instead ofMono, and streaming replacescredentials()withsubscribeToCredentials(onNext, onError)returning aCredentialsSubscriptionhandle.RedisAuthenticationHandlerwires automatic re-auth through the new subscription API (close/replace on resubscribe).TokenBasedRedisCredentialsProvideris reimplemented withCompletableFuture, listener lists, optionalExecutordispatch, and newcreate(..., Executor)factories. Static URI builders, handshake, and Entra ID paths are updated toCompletableFuture; user docs and tests follow the callback model.Reviewed by Cursor Bugbot for commit 18e0954. Bugbot is set up for automated code reviews on this repo. Configure here.