[reactor-optional] Deprecate reactive credentials provider

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

@@ -120,6 +120,7 @@ public static <K, V> RedisAuthenticationHandler<K, V> createDefaultAuthenticatio
      * Each time new credentials are received, the client is re-authenticated. The previous subscription, if any, is disposed of
      * before setting the new subscription.
      */
+    @SuppressWarnings("deprecation")
     public void subscribe() {
         if (credentialsProvider == null || !credentialsProvider.supportsStreaming()) {
             return;

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

@@ -1,10 +1,13 @@
 package io.lettuce.core;
 
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import io.lettuce.core.internal.LettuceAssert;
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import io.lettuce.core.internal.LettuceAssert;
 
 /**
  * Interface for loading {@link RedisCredentials} that are used for authentication. A commonly-used implementation is
@@ -26,9 +29,26 @@ public interface RedisCredentialsProvider {
      * an error occurs during the loading of credentials or credentials could not be found, a runtime exception will be raised.
      *
      * @return a {@link Mono} emitting {@link RedisCredentials} that can be used to authorize a Redis connection.
+     * @deprecated since 7.7, use {@link #resolveCredentialsAsync()} instead. This Reactor-typed method is removed in Lettuce
+     *             8.0, when {@code reactor-core} becomes optional and {@link #resolveCredentialsAsync()} becomes the primary
+     *             credential-resolution contract.
      */
+    @Deprecated
     Mono<RedisCredentials> resolveCredentials();
 
+    /**
+     * Resolve the latest available credentials as a {@link CompletionStage}. This Reactor-free method replaces
+     * {@link #resolveCredentials()} and becomes the primary credential-resolution contract in Lettuce 8.0. Prefer it for new
+     * code; implementations continue to supply credentials through {@link #resolveCredentials()} until 8.0.
+     *
+     * @return a {@link CompletionStage} that completes with the {@link RedisCredentials} used to authorize a Redis connection.
+     * @since 7.7
+     */
+    @SuppressWarnings("deprecation")
+    default CompletionStage<RedisCredentials> resolveCredentialsAsync() {
+        return resolveCredentials().toFuture();
+    }
+
     /**
      * Creates a new {@link RedisCredentialsProvider} from a given {@link Supplier}.
      *
@@ -46,7 +66,7 @@ static RedisCredentialsProvider from(Supplier<RedisCredentials> supplier) {
      * Some implementations of the {@link RedisCredentialsProvider} may support streaming new credentials, based on some event
      * that originates outside the driver. In this case they should indicate that so the {@link RedisAuthenticationHandler} is
      * able to process these new credentials.
-     * 
+     *
      * @return whether the {@link RedisCredentialsProvider} supports streaming credentials.
      */
     default boolean supportsStreaming() {
@@ -65,11 +85,35 @@ default boolean supportsStreaming() {
      *
      * @return a {@link Flux} emitting {@link RedisCredentials}, or throws an exception if streaming is not supported.
      * @throws UnsupportedOperationException if the provider does not support streaming credentials.
+     * @deprecated since 7.7, use {@link #subscribeToCredentials(Consumer, Consumer)} instead; scheduled for removal in Lettuce
+     *             8.0 (when {@code reactor-core} becomes optional).
      */
+    @Deprecated
     default Flux<RedisCredentials> credentials() {
         throw new UnsupportedOperationException("Streaming credentials are not supported by this provider.");
     }
 
+    /**
+     * Subscribe to credential updates produced by this provider. For providers that support streaming (as indicated by
+     * {@link #supportsStreaming()} returning {@code true}), {@code onNext} is invoked whenever new credentials become available
+     * (e.g. token renewal or rotation) and {@code onError} is invoked when the provider observes a failure while producing
+     * credentials. Delivery stops once the returned {@link Subscription} is {@link Subscription#close() closed}.
+     * <p>
+     * Providers that do not support streaming throw an {@link UnsupportedOperationException} by default.
+     *
+     * @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 Subscription} that stops delivery when closed.
+     * @throws UnsupportedOperationException if the provider does not support streaming credentials.
+     * @since 7.7
+     */
+    default 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");
+        Disposable disposable = credentials().subscribe(onNext, onError);
+        return disposable::dispose;
+    }
+
     /**
      * Extension to {@link RedisCredentialsProvider} that resolves credentials immediately without the need to defer the
      * credential resolution.

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/core/RedisCredentialsProviderUnitTests.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2026-Present, Redis Ltd. All rights reserved.
+ * SPDX-License-Identifier: MIT
+ */
+package io.lettuce.core;
+
+import static io.lettuce.TestTags.UNIT_TEST;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+/**
+ * Unit tests for the 7.x forward-compatible additions to {@link RedisCredentialsProvider}: {@code resolveCredentialsAsync()}
+ * and {@code subscribeToCredentials(...)}.
+ *
+ * @author Aleksandar Todorov
+ */
+@SuppressWarnings("deprecation")
+@Tag(UNIT_TEST)
+class RedisCredentialsProviderUnitTests {
+
+    private final RedisCredentials creds = RedisCredentials.just("user", "pass".toCharArray());
+
+    @Test
+    void resolveCredentialsAsyncBridgesResolveCredentials() throws Exception {
+
+        RedisCredentialsProvider provider = () -> Mono.just(creds);
+
+        assertThat(provider.resolveCredentialsAsync().toCompletableFuture().get()).isSameAs(creds);
+    }
+
+    @Test
+    void subscribeToCredentialsDeliversAndCloseStops() {
+
+        Sinks.Many<RedisCredentials> sink = Sinks.many().multicast().directBestEffort();
+        RedisCredentialsProvider provider = new RedisCredentialsProvider() {
+
+            @Override
+            public Mono<RedisCredentials> resolveCredentials() {
+                return sink.asFlux().next();
+            }
+
+            @Override
+            public boolean supportsStreaming() {
+                return true;
+            }
+
+            @Override
+            public Flux<RedisCredentials> credentials() {
+                return sink.asFlux();
+            }
+
+        };
+
+        List<RedisCredentials> received = new ArrayList<>();
+        Subscription subscription = provider.subscribeToCredentials(received::add, t -> {
+        });
+
+        sink.tryEmitNext(creds);
+        assertThat(received).containsExactly(creds);
+
+        subscription.close();
+        sink.tryEmitNext(RedisCredentials.just("user2", "pass2".toCharArray()));
+        assertThat(received).containsExactly(creds); // nothing delivered after close
+    }
+
+}