Shutdown ClusterTopologyRefreshTask properly

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

@@ -26,6 +26,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 
 import io.lettuce.core.ClientOptions;
@@ -64,6 +66,10 @@ class ClusterTopologyRefreshScheduler implements Runnable, ClusterEventListener
 
     private final EventExecutorGroup genericWorkerPool;
 
+    private static final ReentrantLock refreshLock = new ReentrantLock();
+
+    private static final Condition refreshComplete = refreshLock.newCondition();
+
     ClusterTopologyRefreshScheduler(Supplier<ClusterClientOptions> clientOptions, Supplier<Partitions> partitions,
             Supplier<CompletionStage<?>> refreshTopology, ClientResources clientResources) {
 
@@ -112,6 +118,14 @@ public boolean isTopologyRefreshInProgress() {
         return clusterTopologyRefreshTask.get();
     }
 
+    public ReentrantLock getRefreshLock() {
+        return refreshLock;
+    }
+
+    public Condition getRefreshComplete() {
+        return refreshComplete;
+    }
+
     @Override
     public void run() {
 
@@ -323,13 +337,18 @@ private static class ClusterTopologyRefreshTask extends AtomicBoolean implements
 
         public void run() {
 
-            if (compareAndSet(false, true)) {
-                doRun();
-                return;
-            }
-
-            if (logger.isDebugEnabled()) {
-                logger.debug("ClusterTopologyRefreshTask already in progress");
+            refreshLock.lock();
+            try {
+                if (compareAndSet(false, true)) {
+                    doRun();
+                    return;
+                }
+
+                if (logger.isDebugEnabled()) {
+                    logger.debug("ClusterTopologyRefreshTask already in progress");
+                }
+            } finally {
+                refreshLock.unlock();
             }
         }
 
@@ -345,7 +364,13 @@ void doRun() {
                         logger.warn("Cannot refresh Redis Cluster topology", throwable);
                     }
 
-                    set(false);
+                    refreshLock.lock();
+                    try {
+                        set(false);
+                        refreshComplete.signalAll();
+                    } finally {
+                        refreshLock.unlock();
+                    }
                 });
             } catch (Exception e) {
                 logger.warn("Cannot refresh Redis Cluster topology", e);

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

@@ -33,6 +33,8 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -1152,6 +1154,21 @@ public void setPartitions(Partitions partitions) {
     public CompletableFuture<Void> shutdownAsync(long quietPeriod, long timeout, TimeUnit timeUnit) {
 
         suspendTopologyRefresh();
+        ReentrantLock refreshLock = topologyRefreshScheduler.getRefreshLock();
+        Condition refreshComplete = topologyRefreshScheduler.getRefreshComplete();
+
+        refreshLock.lock();
+        try {
+            while (topologyRefreshScheduler.isTopologyRefreshInProgress()) {
+                try {
+                    refreshComplete.await();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+        } finally {
+            refreshLock.unlock();
+        }
 
         return super.shutdownAsync(quietPeriod, timeout, timeUnit);
     }