[reactor-optional] Make clean transition from reactive to async credentials provider

src/main/java/io/lettuce/authx/TokenBasedRedisCredentialsProvider.java

@@ -8,6 +8,7 @@
 
 import io.lettuce.core.RedisCredentials;
 import io.lettuce.core.RedisCredentialsProvider;
+import io.lettuce.core.Subscription;
 import io.lettuce.core.internal.LettuceAssert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -52,7 +53,7 @@ public class TokenBasedRedisCredentialsProvider implements RedisCredentialsProvi
 
     private static final Logger log = LoggerFactory.getLogger(TokenBasedRedisCredentialsProvider.class);
 
-    private static class SimpleSubscription implements CredentialsSubscription {
+    private static class SimpleSubscription implements Subscription {
 
         private final TokenBasedRedisCredentialsProvider provider;
 
@@ -205,7 +206,7 @@ private static void dispatchOnError(SimpleSubscription subscription, Throwable t
      * @return a {@link CompletionStage} that completes with the latest Redis credentials
      */
     @Override
-    public CompletionStage<RedisCredentials> resolveCredentials() {
+    public CompletionStage<RedisCredentials> resolveCredentialsAsync() {
         CompletableFuture<RedisCredentials> result = new CompletableFuture<>();
         credentialsFutureRef.get().whenComplete((creds, throwable) -> {
             if (throwable != null) {
@@ -218,7 +219,7 @@ public CompletionStage<RedisCredentials> resolveCredentials() {
     }
 
     @Override
-    public CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+    public Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
         if (isClosed) {
             throw new IllegalStateException("Credentials provider closed");
         }
@@ -280,7 +281,7 @@ public void close() {
      * <ul>
      * <li>Subscriber {@code onNext}/{@code onError} callbacks for live token renewals run on the {@link TokenManager}'s renewal
      * thread. A slow or blocking subscriber can delay or miss subsequent renewals.</li>
-     * <li>Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future
+     * <li>Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future
      * completes.</li>
      * <li>Replay deliveries to a newly subscribing consumer run on the subscribing thread.</li>
      * </ul>
@@ -323,7 +324,7 @@ public static TokenBasedRedisCredentialsProvider create(TokenAuthConfig tokenAut
      * <ul>
      * <li>Subscriber {@code onNext}/{@code onError} callbacks for live token renewals run on the {@link TokenManager}'s renewal
      * thread. A slow or blocking subscriber can delay or miss subsequent renewals.</li>
-     * <li>Continuations chained off {@link #resolveCredentials()} run on the renewal thread when the initial future
+     * <li>Continuations chained off {@link #resolveCredentialsAsync()} run on the renewal thread when the initial future
      * completes.</li>
      * <li>Replay deliveries to a newly subscribing consumer run on the subscribing thread.</li>
      * </ul>

src/main/java/io/lettuce/core/RedisAuthenticationHandler.java

@@ -6,7 +6,6 @@
  */
 package io.lettuce.core;
 
-import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.event.connection.ReauthenticationEvent;
@@ -48,7 +47,7 @@ public class RedisAuthenticationHandler<K, V> {
 
     private final RedisCredentialsProvider credentialsProvider;
 
-    private final AtomicReference<CredentialsSubscription> credentialsSubscription = new AtomicReference<>();
+    private final AtomicReference<Subscription> credentialsSubscription = new AtomicReference<>();
 
     private final Boolean isPubSubConnection;
 
@@ -128,10 +127,9 @@ public void subscribe() {
             return;
         }
 
-        CredentialsSubscription credentialsSub = credentialsProvider.subscribeToCredentials(this::reauthenticate,
-                this::onError);
+        Subscription credentialsSub = credentialsProvider.subscribeToCredentials(this::reauthenticate, this::onError);
 
-        CredentialsSubscription oldSub = credentialsSubscription.getAndSet(credentialsSub);
+        Subscription oldSub = credentialsSubscription.getAndSet(credentialsSub);
         if (oldSub != null) {
             try {
                 oldSub.close();
@@ -145,7 +143,7 @@ public void subscribe() {
      * Unsubscribes from the current credentials stream.
      */
     public void unsubscribe() {
-        CredentialsSubscription sub = credentialsSubscription.getAndSet(null);
+        Subscription sub = credentialsSubscription.getAndSet(null);
         if (sub != null) {
             try {
                 sub.close();

src/main/java/io/lettuce/core/RedisCredentialsProvider.java

@@ -1,6 +1,5 @@
 package io.lettuce.core;
 
-import java.io.Closeable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
@@ -22,26 +21,17 @@
 @FunctionalInterface
 public interface RedisCredentialsProvider {
 
-    /**
-     * 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();
-
-    }
-
     /**
      * Returns {@link RedisCredentials} that can be used to authorize a Redis connection. Each implementation of
      * {@code RedisCredentialsProvider} can choose its own strategy for loading credentials. For example, an implementation
      * might load credentials from an existing key management system, or load new credentials when credentials are rotated. If
-     * an error occurs during the loading of credentials or credentials could not be found, a runtime exception will be raised.
+     * an error occurs during the loading of credentials or credentials could not be found, the returned {@link CompletionStage}
+     * completes exceptionally.
      *
-     * @return a {@link CompletionStage} emitting {@link RedisCredentials} that can be used to authorize a Redis connection.
+     * @return a {@link CompletionStage} that completes with the {@link RedisCredentials} used to authorize a Redis connection.
+     * @since 7.7
      */
-    CompletionStage<RedisCredentials> resolveCredentials();
+    CompletionStage<RedisCredentials> resolveCredentialsAsync();
 
     /**
      * Creates a new {@link RedisCredentialsProvider} from a given {@link Supplier}.
@@ -101,10 +91,10 @@ default boolean supportsStreaming() {
      *
      * @param onNext consumer invoked with each new {@link RedisCredentials} value, must not be {@code null}.
      * @param onError consumer invoked with errors observed while producing credentials, must not be {@code null}.
-     * @return a {@link CredentialsSubscription} that can be used to stop receiving updates.
+     * @return a {@link Subscription} that can be used to stop receiving updates.
      * @throws UnsupportedOperationException if the provider does not support streaming credentials.
      */
-    default CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+    default Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
         throw new UnsupportedOperationException("Streaming credentials are not supported by this provider.");
     }
 
@@ -116,7 +106,7 @@ default CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials
     interface ImmediateRedisCredentialsProvider extends RedisCredentialsProvider {
 
         @Override
-        default CompletionStage<RedisCredentials> resolveCredentials() {
+        default CompletionStage<RedisCredentials> resolveCredentialsAsync() {
             try {
                 RedisCredentials credentials = resolveCredentialsNow();
                 if (credentials == null) {

src/main/java/io/lettuce/core/RedisHandshake.java

@@ -210,7 +210,8 @@ private CompletableFuture<?> initiateHandshakeResp2(Channel channel, RedisCreden
                     ((RedisCredentialsProvider.ImmediateRedisCredentialsProvider) credentialsProvider).resolveCredentialsNow());
         }
 
-        CompletableFuture<RedisCredentials> credentialsFuture = credentialsProvider.resolveCredentials().toCompletableFuture();
+        CompletableFuture<RedisCredentials> credentialsFuture = credentialsProvider.resolveCredentialsAsync()
+                .toCompletableFuture();
 
         return credentialsFuture.thenComposeAsync(credentials -> dispatchAuthOrPing(channel, credentials));
     }
@@ -243,7 +244,8 @@ private CompletionStage<Map<String, Object>> initiateHandshakeResp3(Channel chan
                     ((RedisCredentialsProvider.ImmediateRedisCredentialsProvider) credentialsProvider).resolveCredentialsNow());
         }
 
-        CompletableFuture<RedisCredentials> credentialsFuture = credentialsProvider.resolveCredentials().toCompletableFuture();
+        CompletableFuture<RedisCredentials> credentialsFuture = credentialsProvider.resolveCredentialsAsync()
+                .toCompletableFuture();
 
         return credentialsFuture.thenComposeAsync(credentials -> dispatchHello(channel, credentials));
     }

src/main/java/io/lettuce/core/RedisURI.java

@@ -976,7 +976,7 @@ private String getAuthority(String scheme, boolean maskCredentials) {
                 // would get asterix for each character of the password.
                 RedisCredentials creds;
                 try {
-                    creds = credentialsProvider.resolveCredentials().toCompletableFuture().join();
+                    creds = credentialsProvider.resolveCredentialsAsync().toCompletableFuture().join();
                 } catch (Exception e) {
                     throw Exceptions.bubble(e);
                 }

src/main/java/io/lettuce/core/StaticCredentialsProvider.java

@@ -44,7 +44,7 @@ public StaticCredentialsProvider(RedisCredentials credentials) {
     }
 
     @Override
-    public CompletionStage<RedisCredentials> resolveCredentials() {
+    public CompletionStage<RedisCredentials> resolveCredentialsAsync() {
         return future;
     }
 

src/main/java/io/lettuce/core/Subscription.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2026-Present, Redis Ltd. All rights reserved.
+ * SPDX-License-Identifier: MIT
+ */
+package io.lettuce.core;
+
+import java.io.Closeable;
+
+/**
+ * Handle to a callback subscription on a Lettuce streaming SPI, such as
+ * {@link io.lettuce.core.RedisCredentialsProvider#subscribeToCredentials(java.util.function.Consumer, java.util.function.Consumer)}
+ * or {@link io.lettuce.core.event.EventBus}. Closing the subscription stops delivery of further values to the registered
+ * callback.
+ * <p>
+ * This is not related to {@code org.reactivestreams.Subscription} or to Redis Pub/Sub channel subscriptions.
+ *
+ * @author Aleksandar Todorov
+ * @since 7.7
+ */
+public interface Subscription extends Closeable {
+
+    /**
+     * Stop delivering to the registered callback. Idempotent; calling it more than once has no further effect and never throws.
+     */
+    @Override
+    void close();
+
+}

src/test/java/io/lettuce/authx/DefaultAzureCredentialsIntegrationTests.java

@@ -79,7 +79,8 @@ public void cleanUp() {
     @Test
     public void azureTokenAuthWithDefaultAzureCredentials() throws ExecutionException, InterruptedException, TimeoutException {
 
-        RedisCredentials credentials = credentialsProvider.resolveCredentials().toCompletableFuture().get(5, TimeUnit.SECONDS);
+        RedisCredentials credentials = credentialsProvider.resolveCredentialsAsync().toCompletableFuture().get(5,
+                TimeUnit.SECONDS);
         assertThat(credentials).isNotNull();
 
         String key = UUID.randomUUID().toString();

src/test/java/io/lettuce/authx/EntraIdIntegrationTests.java

@@ -1,7 +1,6 @@
 package io.lettuce.authx;
 
 import io.lettuce.core.*;
-import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.api.reactive.RedisReactiveCommands;
@@ -116,7 +115,7 @@ public void renewalDuringOperationsTest() throws InterruptedException {
         commandThread.start();
 
         CountDownLatch latch = new CountDownLatch(10); // Wait for at least 10 token renewalss
-        CredentialsSubscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> {
+        Subscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> {
         });
         try {
             assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); // Wait to reach 10 renewals
@@ -150,7 +149,7 @@ public void renewalDuringPubSubOperationsTest() throws InterruptedException {
             pubsubThread.start();
 
             CountDownLatch latch = new CountDownLatch(10);
-            CredentialsSubscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> {
+            Subscription subscription = credentialsProvider.subscribeToCredentials(cred -> latch.countDown(), t -> {
             });
             try {
                 assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); // Wait for at least 10 token renewals

src/test/java/io/lettuce/authx/TokenBasedRedisCredentialsProviderTest.java

@@ -2,7 +2,7 @@
 
 import io.lettuce.TestTags;
 import io.lettuce.core.RedisCredentials;
-import io.lettuce.core.RedisCredentialsProvider.CredentialsSubscription;
+import io.lettuce.core.Subscription;
 import io.lettuce.core.TestTokenManager;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
@@ -44,7 +44,7 @@ public void setUp() {
     public void shouldReturnPreviouslyEmittedTokenWhenResolved() {
         tokenManager.emitToken(testToken("test-user", "token-1"));
 
-        Mono<RedisCredentials> credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentials());
+        Mono<RedisCredentials> credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync());
 
         StepVerifier.create(credentials).assertNext(actual -> {
             assertThat(actual.getUsername()).isEqualTo("test-user");
@@ -57,7 +57,7 @@ public void shouldReturnLatestEmittedTokenWhenResolved() {
         tokenManager.emitToken(testToken("test-user", "token-2"));
         tokenManager.emitToken(testToken("test-user", "token-3")); // Latest token
 
-        Mono<RedisCredentials> credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentials());
+        Mono<RedisCredentials> credentials = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync());
 
         StepVerifier.create(credentials).assertNext(actual -> {
             assertThat(actual.getUsername()).isEqualTo("test-user");
@@ -71,7 +71,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() {
         tokenManager.emitToken(testToken("test-user", "token-1"));
 
         // Test resolveCredentials
-        Mono<RedisCredentials> credentials1 = Mono.fromCompletionStage(credentialsProvider.resolveCredentials());
+        Mono<RedisCredentials> credentials1 = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync());
 
         StepVerifier.create(credentials1).assertNext(actual -> {
             assertThat(actual.getUsername()).isEqualTo("test-user");
@@ -81,7 +81,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() {
         // Emit second token and subscribe another
         tokenManager.emitToken(testToken("test-user", "token-2"));
         tokenManager.emitToken(testToken("test-user", "token-3"));
-        Mono<RedisCredentials> credentials2 = Mono.fromCompletionStage(credentialsProvider.resolveCredentials());
+        Mono<RedisCredentials> credentials2 = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync());
         StepVerifier.create(credentials2).assertNext(actual -> {
             assertThat(actual.getUsername()).isEqualTo("test-user");
             assertThat(new String(actual.getPassword())).isEqualTo("token-3");
@@ -90,7 +90,7 @@ public void shouldReturnTokenEmittedBeforeSubscription() {
 
     @Test
     public void shouldWaitForAndReturnTokenWhenEmittedLater() {
-        Mono<RedisCredentials> result = Mono.fromCompletionStage(credentialsProvider.resolveCredentials());
+        Mono<RedisCredentials> result = Mono.fromCompletionStage(credentialsProvider.resolveCredentialsAsync());
 
         tokenManager.emitTokenWithDelay(testToken("test-user", "delayed-token"), 100); // Emit token after 100ms
         StepVerifier.create(result)
@@ -104,12 +104,12 @@ public void shouldStopDeliveringToSubscribersOnClose() throws InterruptedExcepti
         List<RedisCredentials> received2 = new CopyOnWriteArrayList<>();
         CountDownLatch firstTokenLatch = new CountDownLatch(2);
 
-        CredentialsSubscription sub1 = credentialsProvider.subscribeToCredentials(c -> {
+        Subscription sub1 = credentialsProvider.subscribeToCredentials(c -> {
             received1.add(c);
             firstTokenLatch.countDown();
         }, t -> {
         });
-        CredentialsSubscription sub2 = credentialsProvider.subscribeToCredentials(c -> {
+        Subscription sub2 = credentialsProvider.subscribeToCredentials(c -> {
             received2.add(c);
             firstTokenLatch.countDown();
         }, t -> {
@@ -140,7 +140,7 @@ public void shouldPropagateMultipleTokensOnStream() throws InterruptedException
         List<RedisCredentials> received = new CopyOnWriteArrayList<>();
         CountDownLatch latch = new CountDownLatch(2);
 
-        CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> {
+        Subscription sub = credentialsProvider.subscribeToCredentials(c -> {
             received.add(c);
             latch.countDown();
         }, t -> {
@@ -166,7 +166,7 @@ public void shouldReplayLatestTokenToNewSubscriber() throws InterruptedException
         AtomicReference<RedisCredentials> received = new AtomicReference<>();
         CountDownLatch latch = new CountDownLatch(1);
 
-        CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> {
+        Subscription sub = credentialsProvider.subscribeToCredentials(c -> {
             received.set(c);
             latch.countDown();
         }, t -> {
@@ -187,7 +187,7 @@ public void shouldPropagateTokenRequestErrorsToSubscribers() throws InterruptedE
         CountDownLatch tokensLatch = new CountDownLatch(2);
         CountDownLatch errorLatch = new CountDownLatch(1);
 
-        CredentialsSubscription sub = credentialsProvider.subscribeToCredentials(c -> {
+        Subscription sub = credentialsProvider.subscribeToCredentials(c -> {
             received.add(c);
             tokensLatch.countDown();
         }, t -> {

src/test/java/io/lettuce/core/ConnectionCommandIntegrationTests.java

@@ -285,8 +285,8 @@ void authInvalidPassword() {
         } catch (RedisException e) {
             assertThat(e.getMessage()).startsWith("ERR").contains("AUTH");
             StatefulRedisConnectionImpl<String, String> connectionImpl = (StatefulRedisConnectionImpl<String, String>) connection;
-            assertThat(connectionImpl.getConnectionState().getCredentialsProvider().resolveCredentials().toCompletableFuture()
-                    .join().getPassword()).isNull();
+            assertThat(connectionImpl.getConnectionState().getCredentialsProvider().resolveCredentialsAsync()
+                    .toCompletableFuture().join().getPassword()).isNull();
         } finally {
             connection.close();
         }

src/test/java/io/lettuce/core/MyStreamingRedisCredentialsProvider.java

@@ -30,12 +30,12 @@ public boolean supportsStreaming() {
     }
 
     @Override
-    public CompletionStage<RedisCredentials> resolveCredentials() {
+    public CompletionStage<RedisCredentials> resolveCredentialsAsync() {
         return credentialsFutureRef.get();
     }
 
     @Override
-    public CredentialsSubscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
+    public Subscription subscribeToCredentials(Consumer<RedisCredentials> onNext, Consumer<Throwable> onError) {
         LettuceAssert.notNull(onNext, "onNext consumer must not be null");
         LettuceAssert.notNull(onError, "onError consumer must not be null");
         Listener listener = new Listener(onNext, onError);