Guard AbstractRedisClient.shutdown(…) against races in closing resour…

…ces #1800 shutdown now uses copies of mutable collections to process shutdown signals to avoid concurrent modification exceptions.
redis · Jul 13, 2021 · 9f27e90 · 9f27e90
1 parent a881034
commit 9f27e90
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 4 deletions.
diff --git a/src/main/java/io/lettuce/core/AbstractRedisClient.java b/src/main/java/io/lettuce/core/AbstractRedisClient.java
@@ -546,9 +546,9 @@ public CompletableFuture<Void> shutdownAsync(long quietPeriod, long timeout, Tim
     private CompletableFuture<Void> closeResources() {
 
         List<CompletionStage<Void>> closeFutures = new ArrayList<>();
+        List<Closeable> closeableResources = new ArrayList<>(this.closeableResources);
 
-        while (!closeableResources.isEmpty()) {
-            Closeable closeableResource = closeableResources.iterator().next();
+        for (Closeable closeableResource : closeableResources) {
 
             if (closeableResource instanceof AsyncCloseable) {
 
@@ -560,10 +560,14 @@ private CompletableFuture<Void> closeResources() {
                     logger.debug("Exception on Close: " + e.getMessage(), e);
                 }
             }
-            closeableResources.remove(closeableResource);
+            this.closeableResources.remove(closeableResource);
         }
 
-        for (Channel c : channels) {
+        for (Channel c : channels.toArray(new Channel[0])) {
+
+            if (c == null) {
+                continue;
+            }
 
             ChannelPipeline pipeline = c.pipeline();
 

diff --git a/src/test/java/io/lettuce/core/cluster/RedisClusterClientIntegrationTests.java b/src/test/java/io/lettuce/core/cluster/RedisClusterClientIntegrationTests.java
@@ -431,6 +431,22 @@ void clusterAuth() {
         FastShutdown.shutdown(clusterClient);
     }
 
+    @Test
+    void shutdownWithOpenConnectionShouldCloseCorrectly() {
+
+        ClusterTopologyRefreshOptions refreshOptions = ClusterTopologyRefreshOptions.builder()
+                .enableAllAdaptiveRefreshTriggers().enablePeriodicRefresh(Duration.ofSeconds(1)).build();
+
+        RedisClusterClient clusterClient = RedisClusterClient.create(TestClientResources.get(),
+                RedisURI.Builder.redis(TestSettings.host(), ClusterTestSettings.port1).build());
+
+        clusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(refreshOptions).build());
+
+        clusterClient.connect().sync();
+
+        FastShutdown.shutdown(clusterClient);
+    }
+
     @Test
     void partitionRetrievalShouldFail() {