chore: remove deprecated ClientOptions#cancelCommandsOnReconnectFailure (#3328) (#3346)

* chore: remove deprecated ClientOptions#cancelCommandsOnReconnectFailure
- removed the option from ClientOptions and ClusterClientOptions and related builders
- removed usages from integration tests and unit tests
- removed usage from ReconnectionHandler and DefaultEndpoint

Signed-off-by: The-East-Wind <hariatul1998@gmail.com>

* Also remove the documentation on the cancelCommandsOnReconnectFailure option

---------

Signed-off-by: The-East-Wind <hariatul1998@gmail.com>
Co-authored-by: Tihomir Mateev <tihomir.mateev@gmail.com>

Author: The-East-Wind

docs/advanced-usage.md

@@ -382,26 +382,6 @@ that the command would not be filtered out.</p>
 enabled.</p></td>
 </tr>
 <tr>
-<td>Cancel commands on reconnect failure</td>
-<td><code>cancelCommandsOnReconnectFailure</code></td>
-<td><code>false</code></td>
-</tr>
-<tr>
-<td colspan="3"><p>Since: 3.1, 4.0</p>
-<p><strong>This flag is deprecated and should not be used as it can lead
-to race conditions and protocol offsets. SSL is natively supported by
-Lettuce and does no longer requires the use of SSL tunnels where
-protocol traffic can get out of sync.</strong></p>
-<p>If this flag is <code>true</code> any queued commands will be
-canceled when a reconnect fails within the activation sequence. The
-reconnect itself has two phases: Socket connection and
-protocol/connection activation. In case a connect timeout occurs, a
-connection reset, host lookup fails, this does not affect the
-cancellation of commands. In contrast, where the protocol/connection
-activation fails due to SSL errors or PING before activating connection
-failure, queued commands are canceled.</p></td>
-</tr>
-<tr>
 <td>Policy how to reclaim decode buffer memory</td>
 <td><code>decodeBufferPolicy</code></td>
 <td><code>ratio-based at 75%</code></td>

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

@@ -28,7 +28,6 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
-import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.internal.LettuceAssert;
 import io.lettuce.core.json.JsonParser;
 import io.lettuce.core.json.RedisJsonException;
@@ -56,8 +55,6 @@ public class ClientOptions implements Serializable {
 
     public static final int DEFAULT_BUFFER_USAGE_RATIO = 3;
 
-    public static final boolean DEFAULT_CANCEL_CMD_RECONNECT_FAIL = false;
-
     public static final DisconnectedBehavior DEFAULT_DISCONNECTED_BEHAVIOR = DisconnectedBehavior.DEFAULT;
 
     public static final ReauthenticateBehavior DEFAULT_REAUTHENTICATE_BEHAVIOUR = ReauthenticateBehavior.DEFAULT;
@@ -98,8 +95,6 @@ public class ClientOptions implements Serializable {
 
     private final Predicate<RedisCommand<?, ?, ?>> replayFilter;
 
-    private final boolean cancelCommandsOnReconnectFailure;
-
     private final DecodeBufferPolicy decodeBufferPolicy;
 
     private final DisconnectedBehavior disconnectedBehavior;
@@ -133,7 +128,6 @@ 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;
         this.reauthenticateBehavior = builder.reauthenticateBehavior;
@@ -154,7 +148,6 @@ protected ClientOptions(Builder builder) {
     protected ClientOptions(ClientOptions original) {
         this.autoReconnect = original.isAutoReconnect();
         this.replayFilter = original.getReplayFilter();
-        this.cancelCommandsOnReconnectFailure = original.isCancelCommandsOnReconnectFailure();
         this.decodeBufferPolicy = original.getDecodeBufferPolicy();
         this.disconnectedBehavior = original.getDisconnectedBehavior();
         this.reauthenticateBehavior = original.getReauthenticateBehaviour();
@@ -209,8 +202,6 @@ public static class Builder {
 
         private Predicate<RedisCommand<?, ?, ?>> replayFilter = DEFAULT_REPLAY_FILTER;
 
-        private boolean cancelCommandsOnReconnectFailure = DEFAULT_CANCEL_CMD_RECONNECT_FAIL;
-
         private DecodeBufferPolicy decodeBufferPolicy = DecodeBufferPolicies.ratio(DEFAULT_BUFFER_USAGE_RATIO);
 
         private DisconnectedBehavior disconnectedBehavior = DEFAULT_DISCONNECTED_BEHAVIOR;
@@ -271,23 +262,6 @@ public Builder replayFilter(Predicate<RedisCommand<?, ?, ?>> 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
-         * conditions and protocol offsets. The reason is that it internally calls reset() which causes a protocol offset.</b>
-         * See {@link StatefulConnection#reset}
-         *
-         * @param cancelCommandsOnReconnectFailure true/false
-         * @return {@code this}
-         * @deprecated since 6.2, to be removed with 7.0. This feature is unsafe and may cause protocol offsets if true (i.e.
-         *             Redis commands are completed with previous command values).
-         */
-        @Deprecated
-        public Builder cancelCommandsOnReconnectFailure(boolean cancelCommandsOnReconnectFailure) {
-            this.cancelCommandsOnReconnectFailure = cancelCommandsOnReconnectFailure;
-            return this;
-        }
-
         /**
          * Buffer usage ratio for {@link io.lettuce.core.protocol.CommandHandler}. This ratio controls how often bytes are
          * discarded during decoding. In particular, when buffer usage reaches {@code bufferUsageRatio / bufferUsageRatio + 1}.
@@ -551,8 +525,7 @@ public ClientOptions build() {
     public ClientOptions.Builder mutate() {
         Builder builder = new Builder();
 
-        builder.autoReconnect(isAutoReconnect()).cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure())
-                .replayFilter(getReplayFilter()).decodeBufferPolicy(getDecodeBufferPolicy())
+        builder.autoReconnect(isAutoReconnect()).replayFilter(getReplayFilter()).decodeBufferPolicy(getDecodeBufferPolicy())
                 .disconnectedBehavior(getDisconnectedBehavior()).reauthenticateBehavior(getReauthenticateBehaviour())
                 .readOnlyCommands(getReadOnlyCommands()).publishOnScheduler(isPublishOnScheduler())
                 .pingBeforeActivateConnection(isPingBeforeActivateConnection()).protocolVersion(getConfiguredProtocolVersion())
@@ -586,18 +559,6 @@ public boolean isAutoReconnect() {
         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}.
-     *
-     * @return {@code true} if commands should be cancelled on reconnect failures.
-     * @deprecated since 6.2, to be removed with 7.0. See {@link Builder#cancelCommandsOnReconnectFailure(boolean)}.
-     */
-    @Deprecated
-    public boolean isCancelCommandsOnReconnectFailure() {
-        return cancelCommandsOnReconnectFailure;
-    }
-
     /**
      * Returns the {@link DecodeBufferPolicy} used to reclaim memory.
      *

src/main/java/io/lettuce/core/cluster/ClusterClientOptions.java

@@ -130,9 +130,7 @@ public static ClusterClientOptions.Builder builder(ClientOptions clientOptions)
         }
 
         Builder builder = new Builder();
-        builder.autoReconnect(clientOptions.isAutoReconnect())
-                .cancelCommandsOnReconnectFailure(clientOptions.isCancelCommandsOnReconnectFailure())
-                .decodeBufferPolicy(clientOptions.getDecodeBufferPolicy())
+        builder.autoReconnect(clientOptions.isAutoReconnect()).decodeBufferPolicy(clientOptions.getDecodeBufferPolicy())
                 .disconnectedBehavior(clientOptions.getDisconnectedBehavior())
                 .reauthenticateBehavior(clientOptions.getReauthenticateBehaviour())
                 .pingBeforeActivateConnection(clientOptions.isPingBeforeActivateConnection())
@@ -193,20 +191,6 @@ public Builder bufferUsageRatio(int bufferUsageRatio) {
             return this;
         }
 
-        /**
-         *
-         * @param cancelCommandsOnReconnectFailure true/false
-         * @return
-         * @deprecated since 6.2, to be removed with 7.0. This feature is unsafe and may cause protocol offsets if true (i.e.
-         *             Redis commands are completed with previous command values).
-         */
-        @Override
-        @Deprecated
-        public Builder cancelCommandsOnReconnectFailure(boolean cancelCommandsOnReconnectFailure) {
-            super.cancelCommandsOnReconnectFailure(cancelCommandsOnReconnectFailure);
-            return this;
-        }
-
         @Override
         public Builder decodeBufferPolicy(DecodeBufferPolicy decodeBufferPolicy) {
             super.decodeBufferPolicy(decodeBufferPolicy);
@@ -360,14 +344,14 @@ public ClusterClientOptions.Builder mutate() {
 
         Builder builder = new Builder();
 
-        builder.autoReconnect(isAutoReconnect()).cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure())
-                .decodeBufferPolicy(getDecodeBufferPolicy()).disconnectedBehavior(getDisconnectedBehavior())
-                .reauthenticateBehavior(getReauthenticateBehaviour()).maxRedirects(getMaxRedirects())
-                .publishOnScheduler(isPublishOnScheduler()).pingBeforeActivateConnection(isPingBeforeActivateConnection())
-                .protocolVersion(getConfiguredProtocolVersion()).readOnlyCommands(getReadOnlyCommands())
-                .requestQueueSize(getRequestQueueSize()).scriptCharset(getScriptCharset()).socketOptions(getSocketOptions())
-                .sslOptions(getSslOptions()).suspendReconnectOnProtocolFailure(isSuspendReconnectOnProtocolFailure())
-                .timeoutOptions(getTimeoutOptions()).topologyRefreshOptions(getTopologyRefreshOptions())
+        builder.autoReconnect(isAutoReconnect()).decodeBufferPolicy(getDecodeBufferPolicy())
+                .disconnectedBehavior(getDisconnectedBehavior()).reauthenticateBehavior(getReauthenticateBehaviour())
+                .maxRedirects(getMaxRedirects()).publishOnScheduler(isPublishOnScheduler())
+                .pingBeforeActivateConnection(isPingBeforeActivateConnection()).protocolVersion(getConfiguredProtocolVersion())
+                .readOnlyCommands(getReadOnlyCommands()).requestQueueSize(getRequestQueueSize())
+                .scriptCharset(getScriptCharset()).socketOptions(getSocketOptions()).sslOptions(getSslOptions())
+                .suspendReconnectOnProtocolFailure(isSuspendReconnectOnProtocolFailure()).timeoutOptions(getTimeoutOptions())
+                .topologyRefreshOptions(getTopologyRefreshOptions())
                 .validateClusterNodeMembership(isValidateClusterNodeMembership()).nodeFilter(getNodeFilter());
 
         return builder;

src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java

@@ -495,10 +495,6 @@ public void notifyChannelActive(Channel channel) {
                     logger.debug("{} channelActive() ran into an exception", logPrefix());
                 }
 
-                if (clientOptions.isCancelCommandsOnReconnectFailure()) {
-                    reset();
-                }
-
                 throw e;
             }
         });

src/main/java/io/lettuce/core/protocol/ReconnectionHandler.java

@@ -157,10 +157,6 @@ private void reconnect0(CompletableFuture<Channel> result, SocketAddress remoteA
                         return;
                     }
 
-                    if (clientOptions.isCancelCommandsOnReconnectFailure()) {
-                        connectionFacade.reset();
-                    }
-
                     if (clientOptions.isSuspendReconnectOnProtocolFailure()) {
 
                         logger.error("Disabling autoReconnect due to initialization failure", throwable);

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

@@ -17,7 +17,6 @@
 import io.lettuce.core.protocol.Command;
 import io.lettuce.core.protocol.CommandType;
 import io.lettuce.core.protocol.ProtocolVersion;
-import reactor.core.publisher.Mono;
 
 /**
  * Unit tests for {@link ClientOptions}.
@@ -106,7 +105,6 @@ public JsonValue fromObject(Object object) {
 
     void checkAssertions(ClientOptions sut) {
         assertThat(sut.isAutoReconnect()).isTrue();
-        assertThat(sut.isCancelCommandsOnReconnectFailure()).isFalse();
         assertThat(sut.getProtocolVersion()).isEqualTo(ProtocolVersion.RESP3);
         assertThat(sut.isSuspendReconnectOnProtocolFailure()).isFalse();
         assertThat(sut.getDisconnectedBehavior()).isEqualTo(ClientOptions.DisconnectedBehavior.DEFAULT);

src/test/java/io/lettuce/core/cluster/ClusterClientOptionsUnitTests.java

@@ -41,7 +41,6 @@ void testCopy() {
         assertThat(copy.isValidateClusterNodeMembership()).isEqualTo(options.isValidateClusterNodeMembership());
         assertThat(copy.getRequestQueueSize()).isEqualTo(options.getRequestQueueSize());
         assertThat(copy.isAutoReconnect()).isEqualTo(options.isAutoReconnect());
-        assertThat(copy.isCancelCommandsOnReconnectFailure()).isEqualTo(options.isCancelCommandsOnReconnectFailure());
         assertThat(copy.isSuspendReconnectOnProtocolFailure()).isEqualTo(options.isSuspendReconnectOnProtocolFailure());
         assertThat(copy.getMaxRedirects()).isEqualTo(options.getMaxRedirects());
         assertThat(copy.getScriptCharset()).isEqualTo(StandardCharsets.UTF_8);
@@ -73,8 +72,6 @@ void builderFromDefaultClientOptions() {
         assertThat(clusterClientOptions.getRequestQueueSize()).isEqualTo(clusterClientOptions.getRequestQueueSize());
         assertThat(clusterClientOptions.isAutoReconnect()).isEqualTo(clusterClientOptions.isAutoReconnect());
         assertThat(clusterClientOptions.isCloseStaleConnections()).isEqualTo(clusterClientOptions.isCloseStaleConnections());
-        assertThat(clusterClientOptions.isCancelCommandsOnReconnectFailure())
-                .isEqualTo(clusterClientOptions.isCancelCommandsOnReconnectFailure());
         assertThat(clusterClientOptions.isPublishOnScheduler()).isEqualTo(clusterClientOptions.isPublishOnScheduler());
         assertThat(clusterClientOptions.isSuspendReconnectOnProtocolFailure())
                 .isEqualTo(clusterClientOptions.isSuspendReconnectOnProtocolFailure());
@@ -97,7 +94,6 @@ void builderFromClusterClientOptions() {
         assertThat(copy.isValidateClusterNodeMembership()).isEqualTo(options.isValidateClusterNodeMembership());
         assertThat(copy.getRequestQueueSize()).isEqualTo(options.getRequestQueueSize());
         assertThat(copy.isAutoReconnect()).isEqualTo(options.isAutoReconnect());
-        assertThat(copy.isCancelCommandsOnReconnectFailure()).isEqualTo(options.isCancelCommandsOnReconnectFailure());
         assertThat(copy.isSuspendReconnectOnProtocolFailure()).isEqualTo(options.isSuspendReconnectOnProtocolFailure());
         assertThat(copy.getMaxRedirects()).isEqualTo(options.getMaxRedirects());
         assertThat(copy.getScriptCharset()).isEqualTo(options.getScriptCharset());

src/test/java/io/lettuce/core/protocol/ConnectionFailureIntegrationTests.java

@@ -175,62 +175,6 @@ void failOnReconnectShouldSendEvents() throws Exception {
         }
     }
 
-    /**
-     * Simulates a failure on reconnect by changing the port to a invalid server and triggering a reconnect. Meanwhile a command
-     * is fired to the connection and the watchdog is triggered afterwards to reconnect.
-     *
-     * Expectation: Queued commands are canceled (reset), subsequent commands contain the connection exception.
-     *
-     * @throws Exception
-     */
-    @Test
-    void cancelCommandsOnReconnectFailure() throws Exception {
-
-        client.setOptions(ClientOptions.builder().cancelCommandsOnReconnectFailure(true)
-                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
-
-        RandomResponseServer ts = getRandomResponseServer();
-
-        RedisURI redisUri = RedisURI.create(defaultRedisUri.toURI());
-
-        try {
-            RedisAsyncCommandsImpl<String, String> connection = (RedisAsyncCommandsImpl<String, String>) client
-                    .connect(redisUri).async();
-            ConnectionWatchdog connectionWatchdog = ConnectionTestUtil
-                    .getConnectionWatchdog(connection.getStatefulConnection());
-
-            assertThat(connectionWatchdog.isListenOnChannelInactive()).isTrue();
-
-            connectionWatchdog.setReconnectSuspended(true);
-            redisUri.setPort(TestSettings.nonexistentPort());
-
-            connection.quit();
-            Wait.untilTrue(() -> !connection.getStatefulConnection().isOpen()).waitOrTimeout();
-
-            RedisFuture<String> set1 = connection.set(key, value);
-            RedisFuture<String> set2 = connection.set(key, value);
-
-            assertThat(set1.isDone()).isFalse();
-            assertThat(set1.isCancelled()).isFalse();
-
-            assertThat(connection.getStatefulConnection().isOpen()).isFalse();
-            connectionWatchdog.setReconnectSuspended(false);
-            connectionWatchdog.run(0);
-            Delay.delay(Duration.ofMillis(500));
-            assertThat(connection.getStatefulConnection().isOpen()).isFalse();
-
-            assertThatThrownBy(set1::get).isInstanceOf(CancellationException.class).hasNoCause();
-            assertThatThrownBy(set2::get).isInstanceOf(CancellationException.class).hasNoCause();
-
-            assertThatThrownBy(() -> TestFutures.awaitOrTimeout(connection.info())).isInstanceOf(RedisException.class)
-                    .hasMessageContaining("Invalid first byte");
-
-            connection.getStatefulConnection().close();
-        } finally {
-            ts.shutdown();
-        }
-    }
-
     @Test
     void emitEventOnReconnectFailure() throws Exception {