Merge pull request #3295 from mikawudi/initConnections-connection-leak

fix: connection leak when initConnections

redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java

@@ -56,6 +56,8 @@ public class ClientConnectionsEntry {
 
     private final AtomicLong firstFailTime = new AtomicLong(0);
 
+    private volatile boolean initialized = false;
+
     public ClientConnectionsEntry(RedisClient client, int poolMinSize, int poolMaxSize, int subscribePoolMinSize, int subscribePoolMaxSize,
             ConnectionManager connectionManager, NodeType nodeType) {
         this.client = client;
@@ -81,10 +83,19 @@ public class ClientConnectionsEntry {
                         && connectionManager.getConfig().getReadMode() == ReadMode.MASTER_SLAVE
                             && getNodeType() == NodeType.MASTER;
     }
+
+    public boolean isInitialized() {
+        return this.initialized;
+    }
+
+    public void setInitialized(boolean isInited) {
+        this.initialized = isInited;
+    }
     
     public void setNodeType(NodeType nodeType) {
         this.nodeType = nodeType;
     }
+
     public NodeType getNodeType() {
         return nodeType;
     }
@@ -114,6 +125,9 @@ public class ClientConnectionsEntry {
 
     public void setFreezeReason(FreezeReason freezeReason) {
         this.freezeReason = freezeReason;
+        if (freezeReason != null) {
+            this.initialized = false;
+        }
     }
 
     public FreezeReason getFreezeReason() {

redisson/src/main/java/org/redisson/connection/balancer/LoadBalancerManager.java

@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiConsumer;
 
 import org.redisson.api.NodeType;
 import org.redisson.api.RFuture;
@@ -138,12 +139,25 @@ public class LoadBalancerManager {
 
             if (freezeReason != FreezeReason.RECONNECT
                     || entry.getFreezeReason() == FreezeReason.RECONNECT) {
-                entry.resetFirstFail();
-                entry.setFreezeReason(null);
-                
-                slaveConnectionPool.initConnections(entry);
-                pubSubConnectionPool.initConnections(entry);
-                return true;
+                if (!entry.isInitialized()) {
+                    entry.setInitialized(true);
+                    CountableListener<Void> listener = new CountableListener<Void>() {
+                        @Override
+                        protected void onSuccess(Void value) {
+                            entry.setFreezeReason(null);
+                        }
+                    };
+                    listener.setCounter(2);
+                    BiConsumer<Void, Throwable> initCallBack = (r, ex) -> {
+                        if (ex == null) {
+                            listener.decCounter();
+                        }
+                    };
+                    entry.resetFirstFail();
+                    slaveConnectionPool.initConnections(entry).onComplete(initCallBack);
+                    pubSubConnectionPool.initConnections(entry).onComplete(initCallBack);
+                    return true;
+                }
             }
         }
         return false;