fix: isolate suspect server state per client

Synchronize suspect cleanup with roster refresh and make stale cleanup idempotent.

Author: hhlgt

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -944,7 +944,8 @@ public ObTable addTable(ObServerAddr addr){
             logger.info("server from response not exist in route cache, server ip {}, port {} , execute add Table.", addr.getIp(), addr.getSvrPort());
             ObTable obTable = new ObTable.Builder(addr.getIp(), addr.getSvrPort()) //
                     .setLoginInfo(tenantName, userName, password, database, getClientType(runningMode)) //
-                    .setProperties(getProperties()).setObServerAddr(addr).build();
+                    .setProperties(getProperties()).setObServerAddr(addr)
+                    .setFailureHandler(tableRoute::reportObServerFailure).build();
             tableRoster.put(addr, obTable);
             return obTable;
         } catch (Exception e) {
@@ -983,14 +984,12 @@ public Row transformToRow(String tableName, Object[] rowkey) throws Exception {
     }
 
     public void dealWithRpcTimeoutForSingleTablet(ObServerAddr addr, String tableName, long tabletId) throws Exception {
-        RouteTableRefresher.SuspectObServer suspectAddr = new RouteTableRefresher.SuspectObServer(addr);
-        RouteTableRefresher.addIntoSuspectIPs(suspectAddr);
+        tableRoute.reportObServerFailure(addr);
         tableRoute.refreshPartitionLocation(tableName, tabletId, null);
     }
 
     public void dealWithRpcTimeoutForBatchTablet(ObServerAddr addr, String tableName) throws Exception {
-        RouteTableRefresher.SuspectObServer suspectAddr = new RouteTableRefresher.SuspectObServer(addr);
-        RouteTableRefresher.addIntoSuspectIPs(suspectAddr);
+        tableRoute.reportObServerFailure(addr);
         tableRoute.refreshTabletLocationBatch(tableName);
     }
 

src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableConnection.java

@@ -20,7 +20,6 @@
 import com.alipay.oceanbase.rpc.ObGlobal;
 import com.alipay.oceanbase.rpc.exception.*;
 import com.alipay.oceanbase.rpc.location.LocationUtil;
-import com.alipay.oceanbase.rpc.location.model.RouteTableRefresher;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.login.ObTableLoginRequest;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.login.ObTableLoginResult;
 import com.alipay.oceanbase.rpc.table.ObTable;
@@ -119,9 +118,7 @@ private boolean connect() throws Exception {
 
         if (tries >= maxTryTimes) {
             if (!obTable.isOdpMode() && obTable.getObServerAddr() != null) {
-                RouteTableRefresher.SuspectObServer suspectAddr = new RouteTableRefresher.SuspectObServer(
-                    obTable.getObServerAddr());
-                RouteTableRefresher.addIntoSuspectIPs(suspectAddr);
+                obTable.reportConnectionFailure();
             }
             LOGGER.warn("connect failed after max " + maxTryTimes + " tries "
                         + TraceUtil.formatIpPort(obTable));

src/main/java/com/alipay/oceanbase/rpc/location/model/RouteTableRefresher.java

@@ -22,13 +22,13 @@
 import java.sql.Statement;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.exception.ObTableEntryRefreshException;
 import com.alipay.oceanbase.rpc.exception.ObTableTryLockTimeoutException;
-import com.alipay.oceanbase.rpc.exception.ObTableUnexpectedException;
 import com.alipay.oceanbase.rpc.location.LocationUtil;
 import com.alipay.oceanbase.rpc.table.ObTable;
 import org.slf4j.Logger;
@@ -49,11 +49,15 @@ public class RouteTableRefresher {
 
     private final ScheduledExecutorService                                scheduler = Executors.newScheduledThreadPool(2);
 
-    private final static ConcurrentHashMap<ObServerAddr, Lock>            suspectLocks = new ConcurrentHashMap<>(); // ObServer -> access lock
+    private final ConcurrentHashMap<ObServerAddr, Lock>                   suspectLocks = new ConcurrentHashMap<>(); // ObServer -> access lock
 
-    private final static ConcurrentHashMap<ObServerAddr, SuspectObServer> suspectServers = new ConcurrentHashMap<>(); // ObServer -> information structure
+    private final ConcurrentHashMap<ObServerAddr, SuspectObServer>        suspectServers = new ConcurrentHashMap<>(); // ObServer -> information structure
 
-    private final static HashMap<ObServerAddr, Long>                      serverLastAccessTimestamps = new HashMap<>(); // ObServer -> last access timestamp
+    private final ConcurrentHashMap<ObServerAddr, Long>                   serverLastAccessTimestamps = new ConcurrentHashMap<>(); // ObServer -> last access timestamp
+
+    private final Set<ObServerAddr>                                       activeServers = ConcurrentHashMap.newKeySet();
+
+    private final AtomicBoolean                                           closed = new AtomicBoolean(false);
 
     public RouteTableRefresher(ObTableClient tableClient, ObUserAuth sysUA) {
         this.tableClient = tableClient;
@@ -70,6 +74,9 @@ public void start() {
     }
 
     public void close() {
+        if (!closed.compareAndSet(false, true)) {
+            return;
+        }
         try {
             scheduler.shutdown();
             // wait at most 1 seconds to close the scheduler
@@ -79,6 +86,39 @@ public void close() {
         } catch (InterruptedException e) {
             logger.warn("scheduler await for terminate interrupted: {}.", e.getMessage());
             scheduler.shutdownNow();
+            Thread.currentThread().interrupt();
+        } finally {
+            suspectServers.clear();
+            suspectLocks.clear();
+            serverLastAccessTimestamps.clear();
+            activeServers.clear();
+        }
+    }
+
+    /**
+     * Reconcile the keep-alive state with the latest authoritative tenant roster.
+     * Servers removed from the roster must not remain in, or be re-added to, the suspect set.
+     */
+    public void refreshActiveServers(Collection<ObServerAddr> servers) {
+        if (closed.get()) {
+            return;
+        }
+        Set<ObServerAddr> newServers = new HashSet<>();
+        if (servers != null) {
+            newServers.addAll(servers);
+        }
+        activeServers.retainAll(newServers);
+        activeServers.addAll(newServers);
+
+        for (ObServerAddr addr : suspectServers.keySet()) {
+            if (!newServers.contains(addr)) {
+                removeFromSuspectIPs(addr);
+            }
+        }
+        for (ObServerAddr addr : serverLastAccessTimestamps.keySet()) {
+            if (!newServers.contains(addr)) {
+                serverLastAccessTimestamps.remove(addr);
+            }
         }
     }
 
@@ -127,6 +167,10 @@ private void doRsListCheck() {
     private void doCheckAliveTask() {
         for (Map.Entry<ObServerAddr, SuspectObServer> entry : suspectServers.entrySet()) {
             try {
+                if (!activeServers.contains(entry.getKey())) {
+                    removeFromSuspectIPs(entry.getKey());
+                    continue;
+                }
                 checkAlive(entry.getKey());
             } catch (Exception e) {
                 // silence resolving
@@ -161,7 +205,7 @@ private void checkAlive(ObServerAddr addr) {
             if (t instanceof SQLException) {
                 // occurred during query
                 calcFailureOrClearCache(addr);
-            } if (t instanceof ObTableEntryRefreshException) {
+            } else if (t instanceof ObTableEntryRefreshException) {
                 // occurred during connection construction
                 ObTableEntryRefreshException e = (ObTableEntryRefreshException) t;
                 if (e.isConnectInactive()) {
@@ -192,11 +236,22 @@ private void checkAlive(ObServerAddr addr) {
         }
     }
 
-    public static void addIntoSuspectIPs(SuspectObServer server) throws InterruptedException {
+    public void addIntoSuspectIPs(ObServerAddr addr) {
+        if (addr == null) {
+            return;
+        }
+        addIntoSuspectIPs(new SuspectObServer(addr));
+    }
+
+    private void addIntoSuspectIPs(SuspectObServer server) {
         if (server == null || server.getAddr() == null) {
             return;
         }
         ObServerAddr addr = server.getAddr();
+        if (closed.get() || !activeServers.contains(addr)) {
+            logger.debug("ignore suspect report for inactive server: {}", addr);
+            return;
+        }
         if (suspectServers.get(addr) != null) {
             // already in the list, directly return
             return;
@@ -218,6 +273,9 @@ public static void addIntoSuspectIPs(SuspectObServer server) throws InterruptedE
                         // already in the list, directly break
                         break;
                     }
+                    if (closed.get() || !activeServers.contains(addr)) {
+                        break;
+                    }
                     Long lastServerAccessTs = serverLastAccessTimestamps.get(addr);
                     if (lastServerAccessTs != null) {
                         long interval = System.currentTimeMillis() - lastServerAccessTs;
@@ -235,6 +293,11 @@ public static void addIntoSuspectIPs(SuspectObServer server) throws InterruptedE
                     ++retryTimes;
                     logger.warn("wait to try lock to timeout 1s when add observer into suspect ips, server: {}, tryTimes: {}",
                             addr.toString(), retryTimes, e);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.debug("interrupted while adding observer into suspect ips, server: {}",
+                        addr);
+                    break;
                 }
             } // end while
         } finally {
@@ -247,43 +310,33 @@ public static void addIntoSuspectIPs(SuspectObServer server) throws InterruptedE
     private void removeFromSuspectIPs(ObServerAddr addr) {
         Lock lock = suspectLocks.get(addr);
         if (lock == null) {
-            // lock must have been added before remove
-            throw new ObTableUnexpectedException(String.format("ObServer [%s:%d] need to be add into suspect ips before remove",
-                    addr.getIp(), addr.getSvrPort()));
+            suspectServers.remove(addr);
+            logger.debug("suspect server has already been removed: {}", addr);
+            return;
         }
         boolean acquired = false;
         try {
-            int retryTimes = 0;
-            while (true) {
-                try {
-                    acquired = lock.tryLock(1, TimeUnit.SECONDS);
-                    if (!acquired) {
-                        throw new ObTableTryLockTimeoutException("try to get suspect server lock timeout, timeout: 1s");
-                    }
-                    // no need to remove lock
-                    SuspectObServer server = suspectServers.remove(addr);
-                    if (server != null) {
-                        int failure = server.getFailure();
-                        if (failure < failureLimit) {
-                            ObTable obTable = tableClient.getTableRoute().getTableRoster().getTable(addr);
-                            if (obTable != null && !obTable.isValid()) {
-                                obTable.setValid();
-                            }
-                        }
+            acquired = lock.tryLock(1, TimeUnit.SECONDS);
+            if (!acquired) {
+                logger.debug("defer suspect removal because lock is busy, server: {}", addr);
+                return;
+            }
+            // Keep the lock and cooldown entry until this refresher closes. This prevents a
+            // concurrent add from using a different lock and preserves the existing cooldown.
+            SuspectObServer server = suspectServers.remove(addr);
+            if (server != null) {
+                int failure = server.getFailure();
+                if (failure < failureLimit && activeServers.contains(addr)) {
+                    ObTable obTable = tableClient.getTableRoute().getTableRoster().getTable(addr);
+                    if (obTable != null && !obTable.isValid()) {
+                        obTable.setValid();
                     }
-                    logger.debug("removed server from suspect list: {}", addr);
-                    break;
-                } catch (ObTableTryLockTimeoutException e) {
-                    // if try lock timeout, need to retry
-                    ++retryTimes;
-                    logger.warn("wait to try lock to timeout when add observer into suspect ips, server: {}, tryTimes: {}",
-                            addr.toString(), retryTimes, e);
-                } catch (InterruptedException e) {
-                    // do not throw exception to user layer
-                    // next background task will continue to remove it
-                    logger.warn("waiting to get lock while interrupted by other threads", e);
                 }
             }
+            logger.debug("removed server from suspect list: {}", addr);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.debug("interrupted while removing observer from suspect ips, server: {}", addr);
         } finally {
             if (acquired) {
                 lock.unlock();
@@ -294,6 +347,10 @@ private void removeFromSuspectIPs(ObServerAddr addr) {
     private void calcFailureOrClearCache(ObServerAddr addr) {
         TableRoute tableRoute = tableClient.getTableRoute();
         SuspectObServer server = suspectServers.get(addr);
+        if (server == null) {
+            logger.debug("skip failure calculation for removed suspect server: {}", addr);
+            return;
+        }
         server.incrementFailure();
         int failure = server.getFailure();
         if (failure >= failureLimit) {
@@ -304,6 +361,18 @@ private void calcFailureOrClearCache(ObServerAddr addr) {
                 addr, failure);
     }
 
+    int suspectServerCount() {
+        return suspectServers.size();
+    }
+
+    boolean containsSuspectServer(ObServerAddr addr) {
+        return suspectServers.containsKey(addr);
+    }
+
+    int lastAccessTimestampCount() {
+        return serverLastAccessTimestamps.size();
+    }
+
     public static class SuspectObServer {
         private final  ObServerAddr addr;
         private final long accessTimestamp;

src/main/java/com/alipay/oceanbase/rpc/location/model/TableRoster.java

@@ -19,6 +19,7 @@
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
 
 import com.alipay.oceanbase.rpc.ObTableClient;
 import com.alipay.oceanbase.rpc.exception.ObTableCloseException;
@@ -36,6 +37,7 @@ public class TableRoster {
     private Properties properties = new Properties();
     private Map<String, Object> tableConfigs = new HashMap<>();
     private ObTableClientType clientType;
+    private Consumer<ObServerAddr> failureHandler;
     /*
      * ServerAddr(all) -> ObTableConnection
      */
@@ -65,6 +67,9 @@ public void setProperties(Properties properties) {
     public void setTableConfigs(Map<String, Object> tableConfigs) {
         this.tableConfigs = tableConfigs;
     }
+    public void setFailureHandler(Consumer<ObServerAddr> failureHandler) {
+        this.failureHandler = failureHandler;
+    }
     public ObTable getTable(ObServerAddr addr) {
         return tables.get(addr);
     }
@@ -101,7 +106,8 @@ public List<ObServerAddr> refreshTablesAndGetNewServers(List<ReplicaLocation> ne
 
             ObTable obTable = new ObTable.Builder(addr.getIp(), addr.getSvrPort()) //
                     .setLoginInfo(tenantName, userName, password, database, clientType) //
-                    .setProperties(properties).setConfigs(tableConfigs).setObServerAddr(addr).build();
+                    .setProperties(properties).setConfigs(tableConfigs).setObServerAddr(addr)
+                    .setFailureHandler(failureHandler).build();
             ObTable oldObTable = tables.putIfAbsent(addr, obTable);
             logger.warn("add new table addr, {}", addr.toString());
             if (oldObTable != null) { // maybe create two ob table concurrently, close current ob table

src/main/java/com/alipay/oceanbase/rpc/location/model/TableRoute.java

@@ -64,7 +64,7 @@ public class TableRoute {
     private IndexLocations            indexLocations               = null;                           // global index location
     private TableGroupCache           tableGroupCache              = null;
     private OdpInfo                   odpInfo                      = null;
-    private RouteTableRefresher       routeRefresher               = null;
+    private volatile RouteTableRefresher routeRefresher               = null;
     private long                      lastRefreshMetadataTimestamp = -1;
     public final Lock                 refreshTableRosterLock       = new ReentrantLock();
 
@@ -348,7 +348,8 @@ public void initRoster(TableEntryKey rootServerKey, boolean initialized,
                         tableClient.getPassword(), tableClient.getDatabase(),
                         tableClient.getClientType(runningMode))
                     .setProperties(tableClient.getProperties())
-                    .setConfigs(tableClient.getTableConfigs()).setObServerAddr(addr).build();
+                    .setConfigs(tableClient.getTableConfigs()).setObServerAddr(addr)
+                    .setFailureHandler(this::reportObServerFailure).build();
                 addr2Table.put(addr, obTable);
                 servers.add(addr);
             } catch (Exception e) {
@@ -363,6 +364,7 @@ public void initRoster(TableEntryKey rootServerKey, boolean initialized,
             tableClient.getUserName(), tableClient.getPassword(), tableClient.getDatabase(),
             tableClient.getClientType(runningMode), tableClient.getProperties(),
             tableClient.getTableConfigs());
+        this.tableRoster.setFailureHandler(this::reportObServerFailure);
         this.tableRoster.setTables(addr2Table);
         this.serverRoster.reset(servers);
 
@@ -410,9 +412,17 @@ public void initRoster(TableEntryKey rootServerKey, boolean initialized,
 
     public void launchRouteRefresher() {
         routeRefresher = new RouteTableRefresher(tableClient, sysUA);
+        routeRefresher.refreshActiveServers(serverRoster.getMembers());
         routeRefresher.start();
     }
 
+    public void reportObServerFailure(ObServerAddr addr) {
+        RouteTableRefresher refresher = routeRefresher;
+        if (refresher != null) {
+            refresher.addIntoSuspectIPs(addr);
+        }
+    }
+
     public void removeObServer(ObServerAddr addr) {
         logger.debug("remove useless table addr, {}", addr.toString());
         ConcurrentHashMap<ObServerAddr, ObTable> tables = this.tableRoster.getTables();
@@ -480,6 +490,10 @@ public void refreshRosterByRsList(List<ObServerAddr> newRsList) throws Exception
         // update new ob table and get new server address
         List<ObServerAddr> servers = tableRoster.refreshTablesAndGetNewServers(replicaLocations);
         serverRoster.reset(servers);
+        RouteTableRefresher refresher = routeRefresher;
+        if (refresher != null) {
+            refresher.refreshActiveServers(servers);
+        }
 
         // 2. Get Server LDC info for weak read consistency.
         success = false;