Shutdown ClusterTopologyRefreshTask properly

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

@@ -26,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 
 import io.lettuce.core.ClientOptions;
@@ -64,6 +65,8 @@ class ClusterTopologyRefreshScheduler implements Runnable, ClusterEventListener
 
     private final EventExecutorGroup genericWorkerPool;
 
+    private static final ReentrantLock refreshLock = new ReentrantLock();
+
     ClusterTopologyRefreshScheduler(Supplier<ClusterClientOptions> clientOptions, Supplier<Partitions> partitions,
             Supplier<CompletionStage<?>> refreshTopology, ClientResources clientResources) {
 
@@ -108,27 +111,38 @@ public void suspendTopologyRefresh() {
         }
     }
 
+    /**
+     * @return {@code true} if a topology refresh is in progress.
+     */
     public boolean isTopologyRefreshInProgress() {
         return clusterTopologyRefreshTask.get();
     }
 
+    /**
+     * @return the {@link ReentrantLock} that is used to synchronize topology refresh.
+     */
+    public ReentrantLock getRefreshLock() {
+        return refreshLock;
+    }
+
     @Override
     public void run() {
 
         logger.debug("ClusterTopologyRefreshScheduler.run()");
 
-        if (isEventLoopActive()) {
-
-            if (!clientOptions.get().isRefreshClusterView()) {
-                logger.debug("Periodic ClusterTopologyRefresh is disabled");
-                return;
-            }
-        } else {
+        if (!clientOptions.get().isRefreshClusterView()) {
             logger.debug("Periodic ClusterTopologyRefresh is disabled");
             return;
         }
 
-        clientResources.eventExecutorGroup().submit(clusterTopologyRefreshTask);
+        if (refreshLock.tryLock() && isEventLoopActive()) {
+            clientResources.eventExecutorGroup().submit(clusterTopologyRefreshTask);
+        } else {
+            logger.debug("Event loop is shutting down, skip topology refresh");
+            if (refreshLock.isHeldByCurrentThread()) {
+                refreshLock.unlock();
+            }
+        }
     }
 
     @Override
@@ -338,18 +352,17 @@ void doRun() {
             if (logger.isDebugEnabled()) {
                 logger.debug("ClusterTopologyRefreshTask requesting partitions");
             }
-            try {
-                reloadTopologyAsync.get().whenComplete((ignore, throwable) -> {
 
-                    if (throwable != null) {
-                        logger.warn("Cannot refresh Redis Cluster topology", throwable);
-                    }
+            reloadTopologyAsync.get().whenComplete((ignore, throwable) -> {
 
-                    set(false);
-                });
-            } catch (Exception e) {
-                logger.warn("Cannot refresh Redis Cluster topology", e);
-            }
+                if (throwable != null) {
+                    logger.warn("Cannot refresh Redis Cluster topology", throwable);
+                }
+
+                set(false);
+
+                refreshLock.unlock();
+            });
         }
 
     }

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

@@ -33,6 +33,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -1152,8 +1153,13 @@ public void setPartitions(Partitions partitions) {
     public CompletableFuture<Void> shutdownAsync(long quietPeriod, long timeout, TimeUnit timeUnit) {
 
         suspendTopologyRefresh();
-
-        return super.shutdownAsync(quietPeriod, timeout, timeUnit);
+        ReentrantLock refreshLock = topologyRefreshScheduler.getRefreshLock();
+        refreshLock.lock();
+        return super.shutdownAsync(quietPeriod, timeout, timeUnit).whenComplete((ignore, ex) -> {
+            if (refreshLock.isHeldByCurrentThread()) {
+                refreshLock.unlock();
+            }
+        });
     }
 
     // -------------------------------------------------------------------------

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

@@ -6,6 +6,7 @@
 import static org.mockito.Mockito.*;
 
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -37,11 +38,11 @@ class ClusterTopologyRefreshSchedulerUnitTests {
 
     private ClusterTopologyRefreshScheduler sut;
 
-    private ClusterTopologyRefreshOptions immediateRefresh = ClusterTopologyRefreshOptions.builder()
+    private final ClusterTopologyRefreshOptions immediateRefresh = ClusterTopologyRefreshOptions.builder()
             .enablePeriodicRefresh(1, TimeUnit.MILLISECONDS).enableAllAdaptiveRefreshTriggers().build();
 
-    private ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder().topologyRefreshOptions(immediateRefresh)
-            .build();
+    private final ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
+            .topologyRefreshOptions(immediateRefresh).build();
 
     @Mock
     private ClientResources clientResources;
@@ -78,7 +79,7 @@ void runShouldSubmitRefreshShouldTrigger() {
     void runnableShouldCallPartitionRefresh() {
 
         when(clusterClient.getClusterClientOptions()).thenReturn(clusterClientOptions);
-
+        when(clusterClient.refreshPartitionsAsync()).thenReturn(new CompletableFuture<>());
         when(eventExecutors.submit(any(Runnable.class))).then(invocation -> {
             ((Runnable) invocation.getArguments()[0]).run();
             return null;
@@ -93,6 +94,7 @@ void runnableShouldCallPartitionRefresh() {
     void shouldNotSubmitIfExecutorIsShuttingDown() {
 
         when(eventExecutors.isShuttingDown()).thenReturn(true);
+        when(clusterClient.getClusterClientOptions()).thenReturn(clusterClientOptions);
 
         sut.run();
         verify(eventExecutors, never()).submit(any(Runnable.class));
@@ -102,6 +104,7 @@ void shouldNotSubmitIfExecutorIsShuttingDown() {
     void shouldNotSubmitIfExecutorIsShutdown() {
 
         when(eventExecutors.isShutdown()).thenReturn(true);
+        when(clusterClient.getClusterClientOptions()).thenReturn(clusterClientOptions);
 
         sut.run();
         verify(eventExecutors, never()).submit(any(Runnable.class));
@@ -111,6 +114,7 @@ void shouldNotSubmitIfExecutorIsShutdown() {
     void shouldNotSubmitIfExecutorIsTerminated() {
 
         when(eventExecutors.isTerminated()).thenReturn(true);
+        when(clusterClient.getClusterClientOptions()).thenReturn(clusterClientOptions);
 
         sut.run();
         verify(eventExecutors, never()).submit(any(Runnable.class));