Repository: redis/lettuce (java)

Issue:
Introduce command replay filter to avoid command replaying after reconnect
## Feature Request


#### Is your feature request related to a problem? Please describe
For now, we are using lettuce to handle large scales of traffic. The traffic that comes to REDIS cluster per minute is more than 10 million per minute. In this situation, there are some cases where REDIS went down abruptly with various reasons(hardware failure, instance restart etc). The problem that we are facing is when the restarted REDIS instance is attached to REDIS cluster, there are a big CPU spikes causing latency spikes. It turns out that when REDIS instance goes back to REDIS cluster, lettuce client try to send the failed command to the recovered REDIS INSTANCE using AT LEAST ONCE mode of reliability. After changing auto reconnect mode to false, 
the latency spike problem is gone. But, another problem that we have is we cannot reconnect that  recovered REDIS instance. This is because only way to change reliability is to change auto connect mode. So, my suggestion/question is 
why can't we set the only reliability level regardless of auto connect mode? 
Is there any reason why reliability level should follow the auto reconnect mode? 

A clear and concise description of what the problem is. Ex. I have an issue when [...]

#### Describe the solution you'd like
I want to suggest that if we can adjust the reliability level creating another option like RELIABILITY.
That can solve my problem.

A clear and concise description of what you want to happen. Add any considered drawbacks.
For now, I didn't understand why this RELIABILITY has to follow the auto reconnect mode.




Pull request:
Introduce command replay filter to avoid command replaying after reconnect #1310
<!--
Thank you for proposing a pull request. This template will guide you through the essential steps necessary for a pull request.
-->
Closes #1310

Make sure that:

- [x] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [x] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request.
- [x] You applied code formatting rules using the `mvn formatter:format` target. Don’t submit any formatting related changes.
- [x] You submit test cases (unit or integration tests) that back your changes.

 <!--
Great! Live long and prosper.
-->


Patch:
diff --git a/.github/wordlist.txt b/.github/wordlist.txt
index b4dbd628a..b76feed3e 100644
--- a/.github/wordlist.txt
+++ b/.github/wordlist.txt
@@ -277,4 +277,5 @@ ACR
 AMR
 Entra
 authx
-entraid
\ No newline at end of file
+entraid
+autoReconnect
diff --git a/docs/advanced-usage.md b/docs/advanced-usage.md
index 74f154dee..6b95f7d9a 100644
--- a/docs/advanced-usage.md
+++ b/docs/advanced-usage.md
@@ -327,7 +327,7 @@ client.setOptions(ClientOptions.builder()
 <tbody>
 <tr>
 <td>PING before activating connection</td>
-<td><code>pingBefor eActivateConnection</code></td>
+<td><code>pingBeforeActivateConnection</code></td>
 <td><code>true</code></td>
 </tr>
 <tr>
@@ -362,8 +362,21 @@ queued commands.</p>
 refuse commands and cancel these with an exception.</p></td>
 </tr>
 <tr>
+<td>Replay filter</td>
+<td><code>replayFilter</code></td>
+<td><code>(cmd) -> false</code></td>
+</tr>
+<tr>
+<td colspan="3"><p>Since: 6.6</p>
+<p>Controls which commands are to be filtered out in case the driver
+attempts to reconnect to the server. Returning <code>false</code> means
+that the command would not be filtered out.</p>
+<p>This flag has no effect in case the autoReconnect feature is not
+enabled.</p></td>
+</tr>
+<tr>
 <td>Cancel commands on reconnect failure</td>
-<td><code>cancelCommand sOnReconnectFailure</code></td>
+<td><code>cancelCommandsOnReconnectFailure</code></td>
 <td><code>false</code></td>
 </tr>
 <tr>
@@ -486,7 +499,7 @@ store/trust store.</p></td>
 <tr>
 <td>Timeout Options</td>
 <td><code>timeoutOptions</code></td>
-<td><code>Do n ot timeout commands.</code></td>
+<td><code>Do not timeout commands.</code></td>
 </tr>
 <tr>
 <td colspan="3"><p>Since: 5.1</p>
@@ -550,7 +563,7 @@ client.setOptions(ClusterClientOptions.builder()
 <tbody>
 <tr>
 <td>Periodic cluster topology refresh</td>
-<td><code>en ablePeriodicRefresh</code></td>
+<td><code>enablePeriodicRefresh</code></td>
 <td><code>false</code></td>
 </tr>
 <tr>
@@ -909,6 +922,12 @@ Lettuce because of two reasons:
 2.  Support for using epoll/kqueue with shaded netty requires netty 4.1
     and all parts of netty to be shaded.
 
+!!! NOTE
+    The limitations regarding shaded JAR files are applicable only to 
+    Lettuce versions prior to 5.0.1. Since version 5.0.1, Lettuce no
+    longer releases shaded JAR files, and these limitations are no
+    longer relevant for newer versions.
+
 See also Netty [documentation on native
 transports](http://netty.io/wiki/native-transports.html).
 
@@ -2399,14 +2418,14 @@ independent connections to Redis.
 Lettuce provides two levels of consistency; these are the rules for
 Redis command sends:
 
-Depending on the chosen consistency level:
+#### Depending on the chosen consistency level
 
-- **at-most-once execution**, i. e. no guaranteed execution
+- **at-most-once execution**, i.e. no guaranteed execution
 
-- **at-least-once execution**, i. e. guaranteed execution (with [some
+- **at-least-once execution**, i.e. guaranteed execution (with [some
   exceptions](#exceptions-to-at-least-once))
 
-Always:
+#### Always
 
 - command ordering in the order of invocations
 
@@ -2602,9 +2621,44 @@ re-established, queued commands are re-sent for execution. While a
 connection failure persists, issued commands are buffered.
 
 To change into *at-most-once* consistency level, disable auto-reconnect
-mode. Connections cannot be longer reconnected and thus no retries are
-issued. Not successfully commands are canceled. New commands are
-rejected.
+mode. Connections can no longer be reconnected and thus no retries are
+issued. Unsuccessful commands are canceled. New commands are rejected.
+
+#### Controlling replay of commands in *at-lease-once* mode
+
+!!! NOTE
+    This feature is only available since Lettuce 6.6
+
+One can achieve a more fine-grained control over the commands that are
+replayed after a reconnection by using the option to specify a filter
+predicate. This option is part of the ClientOptions configuration. See 
+[Client Options](advanced-usage.md#client-options) for further reference.
+
+``` java
+Predicate<RedisCommand<?, ?, ?> > filter = cmd -> 
+    cmd.getType().toString().equalsIgnoreCase("DECR");
+
+client.setOptions(ClientOptions.builder()
+    .autoReconnect(true)
+    .replayFilter(filter)
+    .build());
+```
+
+The code above would filter out all `DECR` commands from being replayed
+after a reconnection. Another, perhaps more popular example, would be:
+
+``` java
+Predicate<RedisCommand<?, ?, ?> > filter = cmd -> true;
+
+client.setOptions(ClientOptions.builder()
+    .autoReconnect(true)
+    .replayFilter(filter)
+    .build());
+```
+
+... which disables any command replay, but still allows the driver to
+re-connect, basically providing a way to have auto-reconnect without
+auto-replay of commands.
 
 ### Clustered operations
 
diff --git a/src/main/java/io/lettuce/core/ClientOptions.java b/src/main/java/io/lettuce/core/ClientOptions.java
index 5c02c3b18..a16845b00 100644
--- a/src/main/java/io/lettuce/core/ClientOptions.java
+++ b/src/main/java/io/lettuce/core/ClientOptions.java
@@ -25,6 +25,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.Iterator;
 import java.util.ServiceConfigurationError;
 import java.util.ServiceLoader;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import io.lettuce.core.api.StatefulConnection;
@@ -35,6 +36,7 @@ import io.lettuce.core.protocol.DecodeBufferPolicies;
 import io.lettuce.core.protocol.DecodeBufferPolicy;
 import io.lettuce.core.protocol.ProtocolVersion;
 import io.lettuce.core.protocol.ReadOnlyCommands;
+import io.lettuce.core.protocol.RedisCommand;
 import io.lettuce.core.resource.ClientResources;
 import reactor.core.publisher.Mono;
 
@@ -50,6 +52,8 @@ public class ClientOptions implements Serializable {
 
     public static final boolean DEFAULT_AUTO_RECONNECT = true;
 
+    public static final Predicate<RedisCommand<?, ?, ?>> DEFAULT_REPLAY_FILTER = (cmd) -> false;
+
     public static final int DEFAULT_BUFFER_USAGE_RATIO = 3;
 
     public static final boolean DEFAULT_CANCEL_CMD_RECONNECT_FAIL = false;
@@ -92,6 +96,8 @@ public class ClientOptions implements Serializable {
 
     private final boolean autoReconnect;
 
+    private final Predicate<RedisCommand<?, ?, ?>> replayFilter;
+
     private final boolean cancelCommandsOnReconnectFailure;
 
     private final DecodeBufferPolicy decodeBufferPolicy;
@@ -126,6 +132,7 @@ public class ClientOptions implements Serializable {
 
     protected ClientOptions(Builder builder) {
         this.autoReconnect = builder.autoReconnect;
+        this.replayFilter = builder.replayFilter;
         this.cancelCommandsOnReconnectFailure = builder.cancelCommandsOnReconnectFailure;
         this.decodeBufferPolicy = builder.decodeBufferPolicy;
         this.disconnectedBehavior = builder.disconnectedBehavior;
@@ -146,6 +153,7 @@ public class ClientOptions implements Serializable {
 
     protected ClientOptions(ClientOptions original) {
         this.autoReconnect = original.isAutoReconnect();
+        this.replayFilter = original.getReplayFilter();
         this.cancelCommandsOnReconnectFailure = original.isCancelCommandsOnReconnectFailure();
         this.decodeBufferPolicy = original.getDecodeBufferPolicy();
         this.disconnectedBehavior = original.getDisconnectedBehavior();
@@ -199,6 +207,8 @@ public class ClientOptions implements Serializable {
 
         private boolean autoReconnect = DEFAULT_AUTO_RECONNECT;
 
+        private Predicate<RedisCommand<?, ?, ?>> replayFilter = DEFAULT_REPLAY_FILTER;
+
         private boolean cancelCommandsOnReconnectFailure = DEFAULT_CANCEL_CMD_RECONNECT_FAIL;
 
         private DecodeBufferPolicy decodeBufferPolicy = DecodeBufferPolicies.ratio(DEFAULT_BUFFER_USAGE_RATIO);
@@ -246,6 +256,21 @@ public class ClientOptions implements Serializable {
             return this;
         }
 
+        /**
+         * When {@link #autoReconnect(boolean)} is set to true, this {@link Predicate} is used to filter commands to replay when
+         * the connection is reestablished after a disconnect. Returning <code>false</code> means the command will not be
+         * filtered out and will be replayed. Defaults to replaying all queued commands.
+         *
+         * @param replayFilter a {@link Predicate} to filter commands to replay. Must not be {@code null}.
+         * @see #DEFAULT_REPLAY_FILTER
+         * @return {@code this}
+         * @since 6.6
+         */
+        public Builder replayFilter(Predicate<RedisCommand<?, ?, ?>> replayFilter) {
+            this.replayFilter = replayFilter;
+            return this;
+        }
+
         /**
          * Allows cancelling queued commands in case a reconnect fails.Defaults to {@code false}. See
          * {@link #DEFAULT_CANCEL_CMD_RECONNECT_FAIL}. <b>This flag is deprecated and should not be used as it can lead to race
@@ -527,13 +552,13 @@ public class ClientOptions implements Serializable {
         Builder builder = new Builder();
 
         builder.autoReconnect(isAutoReconnect()).cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure())
-                .decodeBufferPolicy(getDecodeBufferPolicy()).disconnectedBehavior(getDisconnectedBehavior())
-                .reauthenticateBehavior(getReauthenticateBehaviour()).readOnlyCommands(getReadOnlyCommands())
-                .publishOnScheduler(isPublishOnScheduler()).pingBeforeActivateConnection(isPingBeforeActivateConnection())
-                .protocolVersion(getConfiguredProtocolVersion()).requestQueueSize(getRequestQueueSize())
-                .scriptCharset(getScriptCharset()).jsonParser(getJsonParser()).socketOptions(getSocketOptions())
-                .sslOptions(getSslOptions()).suspendReconnectOnProtocolFailure(isSuspendReconnectOnProtocolFailure())
-                .timeoutOptions(getTimeoutOptions());
+                .replayFilter(getReplayFilter()).decodeBufferPolicy(getDecodeBufferPolicy())
+                .disconnectedBehavior(getDisconnectedBehavior()).reauthenticateBehavior(getReauthenticateBehaviour())
+                .readOnlyCommands(getReadOnlyCommands()).publishOnScheduler(isPublishOnScheduler())
+                .pingBeforeActivateConnection(isPingBeforeActivateConnection()).protocolVersion(getConfiguredProtocolVersion())
+                .requestQueueSize(getRequestQueueSize()).scriptCharset(getScriptCharset()).jsonParser(getJsonParser())
+                .socketOptions(getSocketOptions()).sslOptions(getSslOptions())
+                .suspendReconnectOnProtocolFailure(isSuspendReconnectOnProtocolFailure()).timeoutOptions(getTimeoutOptions());
 
         return builder;
     }
@@ -551,6 +576,16 @@ public class ClientOptions implements Serializable {
         return autoReconnect;
     }
 
+    /**
+     * Controls which {@link RedisCommand} will be replayed after a re-connect. The {@link Predicate} returns <code>true</code>
+     * if command should be filtered out and not replayed. Defaults to {@link #DEFAULT_REPLAY_FILTER}.
+     *
+     * @return the currently set {@link Predicate} used to filter out commands to replay
+     */
+    public Predicate<RedisCommand<?, ?, ?>> getReplayFilter() {
+        return replayFilter;
+    }
+
     /**
      * If this flag is {@code true} any queued commands will be canceled when a reconnect fails within the activation sequence.
      * Default is {@code false}.
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
index 502ce4a76..79f2f05f1 100644
--- a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
+++ b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
@@ -33,6 +33,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import io.lettuce.core.ClientOptions;
@@ -81,6 +82,8 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
 
     private final Reliability reliability;
 
+    private final Predicate<RedisCommand<?, ?, ?>> replayFilter;
+
     private final ClientOptions clientOptions;
 
     private final ClientResources clientResources;
@@ -139,6 +142,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         this.clientOptions = clientOptions;
         this.clientResources = clientResources;
         this.reliability = clientOptions.isAutoReconnect() ? Reliability.AT_LEAST_ONCE : Reliability.AT_MOST_ONCE;
+        this.replayFilter = clientOptions.getReplayFilter();
         this.disconnectedBuffer = LettuceFactories.newConcurrentQueue(clientOptions.getRequestQueueSize());
         this.commandBuffer = LettuceFactories.newConcurrentQueue(clientOptions.getRequestQueueSize());
         this.boundedQueues = clientOptions.getRequestQueueSize() != Integer.MAX_VALUE;
@@ -343,6 +347,13 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             return;
         }
 
+        if (replayFilter.test(command)) {
+            if (debugEnabled) {
+                logger.debug("{} writeToDisconnectedBuffer() Filtering out command {}", logPrefix(), command);
+            }
+            return;
+        }
+
         if (debugEnabled) {
             logger.debug("{} writeToDisconnectedBuffer() buffering (disconnected) command {}", logPrefix(), command);
         }
@@ -1033,10 +1044,16 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         private void potentiallyRequeueCommands(Channel channel, RedisCommand<?, ?, ?> sentCommand,
                 Collection<? extends RedisCommand<?, ?, ?>> sentCommands) {
 
+            // do not requeue commands that are done
             if (sentCommand != null && sentCommand.isDone()) {
                 return;
             }
 
+            // do not requeue commands that are to be filtered out
+            if (this.endpoint.replayFilter.test(sentCommand)) {
+                return;
+            }
+
             if (sentCommands != null) {
 
                 boolean foundToSend = false;


Test patch:
diff --git a/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java b/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java
index 12b2cd66c..c11c28c30 100644
--- a/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java
+++ b/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java
@@ -56,6 +56,7 @@ class ClusterNodeEndpointUnitTests {
     @BeforeEach
     void before() {
 
+        when(clientOptions.getReplayFilter()).thenReturn((cmd) -> false);
         when(clientOptions.getRequestQueueSize()).thenReturn(1000);
         when(clientOptions.getDisconnectedBehavior()).thenReturn(ClientOptions.DisconnectedBehavior.DEFAULT);
 
diff --git a/src/test/java/io/lettuce/core/reliability/AtLeastOnceIntegrationTests.java b/src/test/java/io/lettuce/core/reliability/AtLeastOnceIntegrationTests.java
index 7dbac3bf9..df420ce7d 100644
--- a/src/test/java/io/lettuce/core/reliability/AtLeastOnceIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/reliability/AtLeastOnceIntegrationTests.java
@@ -8,8 +8,10 @@ import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 
 import io.lettuce.core.TimeoutOptions;
+import io.lettuce.core.protocol.RedisCommand;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
@@ -372,6 +374,54 @@ class AtLeastOnceIntegrationTests extends AbstractRedisClientTest {
         verificationConnection.getStatefulConnection().close();
     }
 
+    @Test
+    void retryAfterConnectionIsDisconnectedButFiltered() throws Exception {
+        // Do not replay DECR commands after reconnect for some reason
+        Predicate<RedisCommand<?, ?, ?>> filter = cmd -> cmd.getType().toString().equalsIgnoreCase("DECR");
+
+        client.setOptions(ClientOptions.builder().autoReconnect(true).replayFilter(filter)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
+
+        // needs to be increased on slow systems...perhaps...
+        client.setDefaultTimeout(3, TimeUnit.SECONDS);
+
+        StatefulRedisConnection<String, String> connection = client.connect();
+        RedisCommands<String, String> verificationConnection = client.connect().sync();
+
+        connection.sync().set(key, "1");
+
+        ConnectionWatchdog connectionWatchdog = ConnectionTestUtil.getConnectionWatchdog(connection);
+        connectionWatchdog.setListenOnChannelInactive(false);
+
+        connection.async().quit();
+        while (connection.isOpen()) {
+            Delay.delay(Duration.ofMillis(100));
+        }
+
+        assertThat(connection.async().incr(key).await(1, TimeUnit.SECONDS)).isFalse();
+        assertThat(connection.async().decr(key).await(1, TimeUnit.SECONDS)).isFalse();
+        assertThat(connection.async().decr(key).await(1, TimeUnit.SECONDS)).isFalse();
+
+        assertThat(verificationConnection.get("key")).isEqualTo("1");
+
+        assertThat(ConnectionTestUtil.getDisconnectedBuffer(connection).size()).isGreaterThan(0);
+        assertThat(ConnectionTestUtil.getCommandBuffer(connection)).isEmpty();
+
+        connectionWatchdog.setListenOnChannelInactive(true);
+        connectionWatchdog.scheduleReconnect();
+
+        while (!ConnectionTestUtil.getCommandBuffer(connection).isEmpty()
+                || !ConnectionTestUtil.getDisconnectedBuffer(connection).isEmpty()) {
+            Delay.delay(Duration.ofMillis(10));
+        }
+
+        assertThat(connection.sync().get(key)).isEqualTo("2");
+        assertThat(verificationConnection.get(key)).isEqualTo("2");
+
+        connection.close();
+        verificationConnection.getStatefulConnection().close();
+    }
+
     private Throwable getException(RedisFuture<?> command) {
         try {
             command.get();