fix session close race leading to orphan Psub connections

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java

@@ -128,4 +128,6 @@ default boolean supportSentinelBeacon(String clusterName) {
 
     boolean shouldComputeExtraInHash();
 
+    int getSessionRemoveUnusedDelayMillis();
+
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/impl/CommonConfigBean.java

@@ -220,4 +220,10 @@ public long getAbnormalClusterStatusMonitorIntervalMilli() {
                 DEFAULT_ABNORMAL_CLUSTER_STATUS_MONITOR_INTERVAL_MILLI);
     }
 
+    public static final String KEY_SESSION_REMOVE_UNUSED_DELAY_MILLIS = "checker.session.remove.unused.delay.millis";
+
+    public int getSessionRemoveUnusedDelayMillis() {
+        return getIntProperty(KEY_SESSION_REMOVE_UNUSED_DELAY_MILLIS, 3600000);
+    }
+
 }

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/impl/DefaultHealthCheckInstanceFactory.java

@@ -92,6 +92,7 @@ public DefaultHealthCheckInstanceFactory(CheckerConfig checkerConfig, HealthChec
     public void remove(RedisHealthCheckInstance instance) {
         Endpoint endpoint = instance.getEndpoint();
         endpointFactory.remove(new HostPort(endpoint.getHost(), endpoint.getPort()));
+        redisSessionManager.removeSession(endpoint);
         stopCheck(instance);
     }
 

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/session/AbstractInstanceSessionManager.java

@@ -2,6 +2,9 @@
 
 import com.ctrip.xpipe.api.endpoint.Endpoint;
 import com.ctrip.xpipe.api.foundation.FoundationService;
+import com.ctrip.xpipe.api.monitor.EventMonitor;
+import com.ctrip.xpipe.api.monitor.Task;
+import com.ctrip.xpipe.api.monitor.TransactionMonitor;
 import com.ctrip.xpipe.concurrent.AbstractExceptionLogTask;
 import com.ctrip.xpipe.endpoint.HostPort;
 import com.ctrip.xpipe.pool.XpipeNettyClientKeyedObjectPool;
@@ -10,6 +13,7 @@
 import com.ctrip.xpipe.redis.checker.healthcheck.impl.HealthCheckEndpointFactory;
 import com.ctrip.xpipe.redis.core.meta.MetaCache;
 import com.ctrip.xpipe.utils.VisibleForTesting;
+import com.ctrip.xpipe.utils.job.DynamicDelayPeriodTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -64,99 +68,113 @@ public abstract class AbstractInstanceSessionManager implements InstanceSessionM
     private CheckerConfig config;
 
     @VisibleForTesting
-    public static long checkUnusedRedisDelaySeconds = 4;
+    public static long checkRedisDelaySeconds = 4;
+
+    private DynamicDelayPeriodTask removeUnusedTask;
 
 
     @PostConstruct
-    public void postConstruct(){
+    public void postConstruct() {
+        this.removeUnusedTask = new DynamicDelayPeriodTask("RemoveUnusedInstances",
+                this::removeUnusedInstances, config::getSessionRemoveUnusedDelayMillis, scheduled);
+        try {
+            removeUnusedTask.start();
+        } catch (Exception e) {
+            logger.error("[postConstruct] start removeUnusedTask fail", e);
+            EventMonitor.DEFAULT.logAlertEvent("session-removeUnused-startFail");
+        }
+
         scheduled.scheduleAtFixedRate(new AbstractExceptionLogTask() {
             @Override
-            protected void doRun() throws Exception {
-                try {
-                    removeUnusedInstances();
-                } catch (Exception e) {
-                    logger.error("[removeUnusedInstances]", e);
-                }
-
+            protected void doRun() {
                 for(RedisSession redisSession : sessions.values()){
-                    try{
+                    try {
                         redisSession.check();
-                    }catch (Exception e){
+                    } catch (Exception e) {
                         logger.error("[check]" + redisSession, e);
                     }
                 }
             }
-        }, checkUnusedRedisDelaySeconds, checkUnusedRedisDelaySeconds, TimeUnit.SECONDS);
+        }, checkRedisDelaySeconds, checkRedisDelaySeconds, TimeUnit.SECONDS);
     }
 
     @Override
-    public RedisSession findOrCreateSession(Endpoint endpoint) {
+    public synchronized RedisSession findOrCreateSession(Endpoint endpoint) {
         RedisSession session = sessions.get(endpoint);
 
         if (session == null) {
-            synchronized (this) {
-                session = sessions.get(endpoint);
-                if (session == null) {
-                    session = new RedisSession(endpoint, scheduled, keyedObjectPool, config);
-                    sessions.put(endpoint, session);
-                }
-            }
+            session = new RedisSession(endpoint, scheduled, keyedObjectPool, config);
+            sessions.put(endpoint, session);
         }
 
         return session;
     }
 
     @Override
-    public RedisSession findOrCreateSession(HostPort hostPort) {
+    public synchronized RedisSession findOrCreateSession(HostPort hostPort) {
         return findOrCreateSession(endpointFactory.getOrCreateEndpoint(hostPort));
     }
 
     @VisibleForTesting
-    protected void removeUnusedInstances() {
-        Set<Endpoint> currentStoredRedises = sessions.keySet();
-        if(currentStoredRedises.isEmpty())
-            return;
-
-        Set<HostPort> redisInUse = getInUseInstances();
-        if(redisInUse == null || redisInUse.isEmpty()) {
-            return;
-        }
-        List<Endpoint> unusedRedises = new LinkedList<>();
-
-        for(Endpoint endpoint : currentStoredRedises) {
-            if(!redisInUse.contains(new HostPort(endpoint.getHost(), endpoint.getPort()))) {
-                unusedRedises.add(endpoint);
-            }
-        }
+    protected synchronized void removeUnusedInstances() {
+        TransactionMonitor.DEFAULT.logTransactionSwallowException(
+                "session.cleanup", "removeUnusedInstances", new Task() {
+            @Override
+            public void go() {
+                Set<Endpoint> currentStoredRedises = sessions.keySet();
+                if(currentStoredRedises.isEmpty())
+                    return;
+
+                Set<HostPort> redisInUse = getInUseInstances();
+                if(redisInUse == null || redisInUse.isEmpty()) {
+                    return;
+                }
+                List<Endpoint> unusedRedises = new LinkedList<>();
 
-        if(unusedRedises.isEmpty()) {
-            return;
-        }
-        unusedRedises.forEach(endpoint -> {
-            RedisSession redisSession = sessions.getOrDefault(endpoint, null);
-            if(redisSession != null) {
-                logger.info("[removeUnusedRedises]Redis: {} not in use, remove from session manager", endpoint);
-                // add try logic to continue working on others
-                try {
-                    redisSession.closeConnection();
-                } catch (Exception ignore) {
+                for(Endpoint endpoint : currentStoredRedises) {
+                    if(!redisInUse.contains(new HostPort(endpoint.getHost(), endpoint.getPort()))) {
+                        unusedRedises.add(endpoint);
+                    }
+                }
 
+                if(unusedRedises.isEmpty()) {
+                    return;
                 }
-                sessions.remove(endpoint);
+                unusedRedises.forEach(endpoint -> {
+                    try {
+                        logger.info("[removeUnusedRedises]Redis: {} not in use, remove from session manager", endpoint);
+                        removeSession(endpoint);
+                    } catch (Exception e) {
+                        logger.warn("[removeUnusedRedises] close session {} failed", endpoint, e);
+                    }
+                });
+            }
+
+            @Override
+            public java.util.Map<String, Object> getData() {
+                return null;
             }
         });
     }
 
     protected abstract Set<HostPort> getInUseInstances();
 
 
+    public synchronized boolean removeSession(Endpoint endpoint) {
+        RedisSession session = sessions.remove(endpoint);
+        if (session == null) return false;
+        session.close();
+        return true;
+    }
+
+
     protected void closeAllConnections() {
         try {
             executors.execute(new Runnable() {
                 @Override
                 public void run() {
                     for (RedisSession session : sessions.values()) {
-                        session.closeConnection();
+                        session.close();
                     }
                 }
             });
@@ -167,6 +185,13 @@ public void run() {
 
     @PreDestroy
     public void preDestroy(){
+        if (removeUnusedTask != null) {
+            try {
+                removeUnusedTask.stop();
+            } catch (Exception e) {
+                logger.error("[preDestroy] stop removeUnusedTask fail", e);
+            }
+        }
         closeAllConnections();
     }
 

redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/session/InstanceSessionManager.java

@@ -13,4 +13,6 @@ public interface InstanceSessionManager {
     RedisSession findOrCreateSession(Endpoint endpoint);
 
     RedisSession findOrCreateSession(HostPort hostPort);
+
+    boolean removeSession(Endpoint endpoint);
 }