Relax timeouts on MOVING rebind without buffered commands

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

@@ -154,15 +154,26 @@ private void rebind(MovingEvent movingEvent) {
         channel.attr(REBIND_ATTRIBUTE).set(RebindState.STARTED);
         rebindAwareAddressSupplier.rebind(movingEvent.getTime(), movingEvent.getEndpoint());
 
+        // Relax the timeouts for the whole re-bind window, not only while commands are still in flight. While the
+        // re-bind is in progress the endpoint reports itself as disconnected (see DefaultEndpoint#isConnected), so
+        // every command issued after the MOVING notification is buffered until the connection to the new endpoint has
+        // been established. Those commands need the relaxed timeout just as much as the ones already on the wire.
+        notifyRebindStarted(movingEvent.getTime(), movingEvent.getEndpoint());
+
         ChannelPipeline pipeline = channel.pipeline();
         CommandHandler commandHandler = pipeline.get(CommandHandler.class);
         if (commandHandler.getStack().isEmpty()) {
             logger.debug("[{}] Closing channel as part of rebind", ChannelLogDescriptor.logDescriptor(channel));
             channel.close().awaitUninterruptibly();
             channel.attr(REBIND_ATTRIBUTE).set(RebindState.COMPLETED);
-        } else {
-            notifyRebindStarted(movingEvent.getTime(), movingEvent.getEndpoint());
+
+            // The channel is closed already, so channelReadComplete() is not going to fire for it again and cannot
+            // report the completion. Report it here, which lifts the relaxed timeouts after the grace period.
+            notifyRebindCompleted();
         }
+
+        // Otherwise CommandHandler#decode flips the state to COMPLETED once the stack has been drained and
+        // channelReadComplete() closes the channel and reports the completion of the re-bind.
     }
 
     private String getMigratingShards(PushMessage message) {

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

@@ -247,7 +247,11 @@ void testOnPushMessageMovingWithEmptyStack() {
         verify(rebindAwareAddressSupplier).rebind(eq(Duration.ofSeconds(15)), eq(new InetSocketAddress("127.0.0.1", 6380)));
         verify(channel).close();
         verify(rebindAttribute).set(RebindState.COMPLETED);
-        verify(component1, never()).onRebindStarted(any(), any()); // Not called when stack is empty
+        // Commands issued after the MOVING notification are buffered until the new connection is up, so the timeouts
+        // have to be relaxed for the re-bind window even when nothing was in flight when the notification arrived.
+        verify(component1).onRebindStarted(eq(Duration.ofSeconds(15)), eq(new InetSocketAddress("127.0.0.1", 6380)));
+        // The channel is closed here, so this is the only place that can report the completion of the re-bind.
+        verify(component1).onRebindCompleted();
     }
 
     @Test
@@ -275,6 +279,8 @@ void testOnPushMessageMovingWithNonEmptyStack() {
         verify(rebindAttribute).set(RebindState.STARTED);
         verify(channel, never()).close();
         verify(component1).onRebindStarted(any(), any()); // Called when stack is not empty
+        // The stack still has to be drained, channelReadComplete() reports the completion later on.
+        verify(component1, never()).onRebindCompleted();
     }
 
     @Test
@@ -643,6 +649,8 @@ void testMovingEventWithNullEndpoint() {
         verify(rebindAttribute).set(RebindState.STARTED);
         verify(channel, never()).close();
         verify(component1).onRebindStarted(any(), any()); // Called when stack is not empty
+        // The stack still has to be drained, channelReadComplete() reports the completion later on.
+        verify(component1, never()).onRebindCompleted();
     }
 
     /**