Add metrics for tracking total disconnected time and reconnection attempts

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

@@ -52,6 +52,8 @@ public class ClientOptions implements Serializable {
 
     public static final boolean DEFAULT_AUTO_RECONNECT = true;
 
+    public static final boolean DEFAULT_PROACTIVE_REBIND = false;
+
     public static final Predicate<RedisCommand<?, ?, ?>> DEFAULT_REPLAY_FILTER = (cmd) -> false;
 
     public static final int DEFAULT_BUFFER_USAGE_RATIO = 3;
@@ -96,6 +98,8 @@ public class ClientOptions implements Serializable {
 
     private final boolean autoReconnect;
 
+    private final boolean proactiveRebind;
+
     private final Predicate<RedisCommand<?, ?, ?>> replayFilter;
 
     private final boolean cancelCommandsOnReconnectFailure;
@@ -132,6 +136,7 @@ public class ClientOptions implements Serializable {
 
     protected ClientOptions(Builder builder) {
         this.autoReconnect = builder.autoReconnect;
+        this.proactiveRebind = builder.proactiveRebind;
         this.replayFilter = builder.replayFilter;
         this.cancelCommandsOnReconnectFailure = builder.cancelCommandsOnReconnectFailure;
         this.decodeBufferPolicy = builder.decodeBufferPolicy;
@@ -153,6 +158,7 @@ protected ClientOptions(Builder builder) {
 
     protected ClientOptions(ClientOptions original) {
         this.autoReconnect = original.isAutoReconnect();
+        this.proactiveRebind = original.isProactiveRebindEnabled();
         this.replayFilter = original.getReplayFilter();
         this.cancelCommandsOnReconnectFailure = original.isCancelCommandsOnReconnectFailure();
         this.decodeBufferPolicy = original.getDecodeBufferPolicy();
@@ -207,6 +213,8 @@ public static class Builder {
 
         private boolean autoReconnect = DEFAULT_AUTO_RECONNECT;
 
+        private boolean proactiveRebind = DEFAULT_PROACTIVE_REBIND;
+
         private Predicate<RedisCommand<?, ?, ?>> replayFilter = DEFAULT_REPLAY_FILTER;
 
         private boolean cancelCommandsOnReconnectFailure = DEFAULT_CANCEL_CMD_RECONNECT_FAIL;
@@ -256,6 +264,20 @@ public Builder autoReconnect(boolean autoReconnect) {
             return this;
         }
 
+        /**
+         * Configure whether the driver should listen for server events that indicate the current endpoint is being re-bound.
+         * When enabled, the proactive re-bind will help with the connection handover and reduce the number of failed commands.
+         * This feature requires the server to support proactive re-binds. Defaults to {@code false}. See
+         * {@link #DEFAULT_PROACTIVE_REBIND}.
+         *
+         * @param proactiveRebind true/false
+         * @return {@code this}
+         */
+        public Builder proactiveRebind(boolean proactiveRebind) {
+            this.proactiveRebind = proactiveRebind;
+            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
@@ -551,14 +573,15 @@ public ClientOptions build() {
     public ClientOptions.Builder mutate() {
         Builder builder = new Builder();
 
-        builder.autoReconnect(isAutoReconnect()).cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure())
-                .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());
+        builder.autoReconnect(isAutoReconnect()).proactiveRebind(isProactiveRebindEnabled())
+                .cancelCommandsOnReconnectFailure(isCancelCommandsOnReconnectFailure()).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;
     }
@@ -576,6 +599,19 @@ public boolean isAutoReconnect() {
         return autoReconnect;
     }
 
+    /**
+     * Controls auto-reconnect behavior on connections. If auto-reconnect is {@code true} (default), it is enabled. As soon as a
+     * connection gets closed/reset without the intention to close it, the client will try to reconnect and re-issue any queued
+     * commands.
+     *
+     * This flag has also the effect that disconnected connections will refuse commands and cancel these with an exception.
+     *
+     * @return {@code true} if auto-reconnect is enabled.
+     */
+    public boolean isProactiveRebindEnabled() {
+        return proactiveRebind;
+    }
+
     /**
      * 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}.

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

@@ -27,6 +27,7 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import io.lettuce.core.protocol.RebindAwareConnectionWatchdog;
 import jdk.net.ExtendedSocketOptions;
 import reactor.core.publisher.Mono;
 import io.lettuce.core.internal.LettuceAssert;
@@ -153,9 +154,16 @@ protected ConnectionWatchdog createConnectionWatchdog() {
         LettuceAssert.assertState(bootstrap != null, "Bootstrap must be set for autoReconnect=true");
         LettuceAssert.assertState(socketAddressSupplier != null, "SocketAddressSupplier must be set for autoReconnect=true");
 
-        ConnectionWatchdog watchdog = new ConnectionWatchdog(clientResources.reconnectDelay(), clientOptions, bootstrap,
-                clientResources.timer(), clientResources.eventExecutorGroup(), socketAddressSupplier, reconnectionListener,
-                connection, clientResources.eventBus(), endpoint);
+        ConnectionWatchdog watchdog;
+        if (clientOptions.isProactiveRebindEnabled()) {
+            watchdog = new RebindAwareConnectionWatchdog(clientResources.reconnectDelay(), clientOptions, bootstrap,
+                    clientResources.timer(), clientResources.eventExecutorGroup(), socketAddressSupplier, reconnectionListener,
+                    connection, clientResources.eventBus(), endpoint, clientResources.connectionMonitor());
+        } else {
+            watchdog = new ConnectionWatchdog(clientResources.reconnectDelay(), clientOptions, bootstrap,
+                    clientResources.timer(), clientResources.eventExecutorGroup(), socketAddressSupplier, reconnectionListener,
+                    connection, clientResources.eventBus(), endpoint, clientResources.connectionMonitor());
+        }
 
         endpoint.registerConnectionWatchdog(watchdog);
 

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

@@ -280,7 +280,7 @@ private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandalone
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
-            writer = new CommandExpiryWriter(writer, getOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -413,7 +413,7 @@ private <K, V> ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectPubS
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
-            writer = new CommandExpiryWriter(writer, getOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -580,7 +580,7 @@ private <K, V> ConnectionFuture<StatefulRedisSentinelConnection<K, V>> doConnect
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getOptions())) {
-            writer = new CommandExpiryWriter(writer, getOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {

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

@@ -24,18 +24,26 @@
 @SuppressWarnings("serial")
 public class TimeoutOptions implements Serializable {
 
+    public static final Duration DISABLED_TIMEOUT = Duration.ZERO.minusSeconds(1);
+
     public static final boolean DEFAULT_TIMEOUT_COMMANDS = false;
 
+    public static final Duration DEFAULT_RELAXED_TIMEOUT = DISABLED_TIMEOUT;
+
     private final boolean timeoutCommands;
 
     private final boolean applyConnectionTimeout;
 
+    private final Duration relaxedTimeout;
+
     private final TimeoutSource source;
 
-    private TimeoutOptions(boolean timeoutCommands, boolean applyConnectionTimeout, TimeoutSource source) {
+    private TimeoutOptions(boolean timeoutCommands, boolean applyConnectionTimeout, TimeoutSource source,
+            Duration relaxedTimeout) {
 
         this.timeoutCommands = timeoutCommands;
         this.applyConnectionTimeout = applyConnectionTimeout;
+        this.relaxedTimeout = relaxedTimeout;
         this.source = source;
     }
 
@@ -84,6 +92,8 @@ public static class Builder {
 
         private boolean applyConnectionTimeout = false;
 
+        private Duration relaxedTimeout = DEFAULT_RELAXED_TIMEOUT;
+
         private TimeoutSource source;
 
         /**
@@ -107,6 +117,25 @@ public Builder timeoutCommands(boolean enabled) {
             return this;
         }
 
+        /**
+         * Enable proactive timeout relaxing. Disabled by default, see {@link #DEFAULT_RELAXED_TIMEOUT}.
+         * <p/>
+         * If the Redis server supports this, and the client is set up to use it by the
+         * {@link ClientOptions#isProactiveRebindEnabled()} option, the client would listen to notifications that the current
+         * endpoint is about to go down (as part of some maintenance activity, for example). In such cases, the driver could
+         * extend the existing timeout settings for newly issued commands, or such that are in flight, to make sure they do not
+         * time out during this process. These commands could be either a part of the offline buffer or waiting for a reply.
+         *
+         * @param duration {@link Duration} to relax timeouts proactively, must not be {@code null}.
+         * @return {@code this}
+         */
+        public Builder proactiveTimeoutsRelaxing(Duration duration) {
+            LettuceAssert.notNull(duration, "Duration must not be null");
+
+            this.relaxedTimeout = duration;
+            return this;
+        }
+
         /**
          * Set a fixed timeout for all commands.
          *
@@ -158,7 +187,7 @@ public TimeoutOptions build() {
                 }
             }
 
-            return new TimeoutOptions(timeoutCommands, applyConnectionTimeout, source);
+            return new TimeoutOptions(timeoutCommands, applyConnectionTimeout, source, relaxedTimeout);
         }
 
     }
@@ -177,6 +206,13 @@ public boolean isApplyConnectionTimeout() {
         return applyConnectionTimeout;
     }
 
+    /**
+     * @return the {@link Duration} to relax timeouts proactively, {@link #DISABLED_TIMEOUT} if disabled.
+     */
+    public Duration getRelaxedTimeout() {
+        return relaxedTimeout;
+    }
+
     /**
      * @return the timeout source to determine the timeout for a {@link RedisCommand}. Can be {@code null} if
      *         {@link #isTimeoutCommands()} is {@code false}.

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

@@ -548,7 +548,7 @@ <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectToNodeAsync(RedisC
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getClusterClientOptions())) {
-            writer = new CommandExpiryWriter(writer, getClusterClientOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getClusterClientOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -634,7 +634,7 @@ <K, V> ConnectionFuture<StatefulRedisPubSubConnection<K, V>> connectPubSubToNode
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getClusterClientOptions())) {
-            writer = new CommandExpiryWriter(writer, getClusterClientOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getClusterClientOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -680,7 +680,7 @@ private <K, V> CompletableFuture<StatefulRedisClusterConnection<K, V>> connectCl
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getClusterClientOptions())) {
-            writer = new CommandExpiryWriter(writer, getClusterClientOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getClusterClientOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -798,7 +798,7 @@ private <K, V> CompletableFuture<StatefulRedisClusterPubSubConnection<K, V>> con
         RedisChannelWriter writer = endpoint;
 
         if (CommandExpiryWriter.isSupported(getClusterClientOptions())) {
-            writer = new CommandExpiryWriter(writer, getClusterClientOptions(), getResources());
+            writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getClusterClientOptions(), getResources());
         }
 
         if (CommandListenerWriter.isSupported(getCommandListeners())) {
@@ -1091,6 +1091,7 @@ private CompletionStage<Partitions> fetchPartitions(Iterable<RedisURI> topologyR
                 getClusterClientOptions().getSocketOptions().getConnectTimeout(), useDynamicRefreshSources());
 
         return topology.thenApply(partitions -> {
+            logger.debug("Topology Refresh Views: {} ", partitions);
 
             if (partitions.isEmpty()) {
                 throw new RedisException(String.format("Cannot retrieve initial cluster partitions from initial URIs %s",
@@ -1108,7 +1109,7 @@ private CompletionStage<Partitions> fetchPartitions(Iterable<RedisURI> topologyR
             }
 
             topologyRefreshScheduler.activateTopologyRefreshIfNeeded();
-
+            logger.debug("Topology Refresh loadedPartitions: {}", loadedPartitions);
             return loadedPartitions;
         });
     }

src/main/java/io/lettuce/core/metrics/ConnectionMonitor.java

@@ -0,0 +1,33 @@
+package io.lettuce.core.metrics;
+
+public interface ConnectionMonitor {
+
+    static ConnectionMonitor disabled() {
+
+        return new ConnectionMonitor() {
+
+            @Override
+            public void recordDisconnectedTime(String epid, long time) {
+
+            }
+
+            @Override
+            public void incrementReconnectionAttempts(String epid) {
+
+            }
+
+            @Override
+            public boolean isEnabled() {
+                return false;
+            }
+
+        };
+    }
+
+    void recordDisconnectedTime(String epid, long time);
+
+    void incrementReconnectionAttempts(String epid);
+
+    boolean isEnabled();
+
+}