fix getAllCells when initial (#157)

src/main/java/com/jd/jdbc/discovery/TopologyWatcherManager.java

@@ -18,20 +18,11 @@
 
 package com.jd.jdbc.discovery;
 
-import com.jd.jdbc.common.util.CollectionUtils;
 import com.jd.jdbc.context.IContext;
-import com.jd.jdbc.sqlparser.support.logging.Log;
-import com.jd.jdbc.sqlparser.support.logging.LogFactory;
-import com.jd.jdbc.topo.TopoException;
 import com.jd.jdbc.topo.TopoServer;
-import com.jd.jdbc.util.threadpool.VtThreadFactoryBuilder;
-import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -43,36 +34,30 @@ public enum TopologyWatcherManager {
 
     private Map<String, TopologyWatcher> cellTopologyWatcherMap = null;
 
-    private Map<String, Set<String>> globalKeyspacesMap = null;
-
     private final Lock lock = new ReentrantLock();
 
-    private static final Log LOGGER = LogFactory.getLog(TopologyWatcherManager.class);
-
-    private ScheduledThreadPoolExecutor scheduledExecutor;
-
     TopologyWatcherManager() {
         cellTopologyWatcherMap = new ConcurrentHashMap<>(16);
-        globalKeyspacesMap = new ConcurrentHashMap<>(16);
-
-        scheduledExecutor = new ScheduledThreadPoolExecutor(1, new VtThreadFactoryBuilder.DefaultThreadFactory("reload-cell-schedule", true));
-        scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
-        scheduledExecutor.setRemoveOnCancelPolicy(true);
     }
 
-    public void startWatch(IContext ctx, TopoServer topoServer, String cell, String tabletKeyspace, TimeUnit timeUnit) {
+    public void startWatch(IContext ctx, TopoServer topoServer, String cell, String tabletKeyspace) {
         lock.lock();
         try {
-            String serverAddress = topoServer.getServerAddress();
-            if (!globalKeyspacesMap.containsKey(serverAddress)) {
-                globalKeyspacesMap.put(serverAddress, new HashSet<>());
-
-                startTickerReloadCell(ctx, topoServer, timeUnit);
+            if (!cellTopologyWatcherMap.containsKey(cell)) {
+                TopologyWatcher topologyWatcher = new TopologyWatcher(topoServer, cell, tabletKeyspace);
+                topologyWatcher.start(ctx);
+                cellTopologyWatcherMap.put(cell, topologyWatcher);
             }
-            globalKeyspacesMap.get(serverAddress).add(tabletKeyspace);
+        } finally {
+            lock.unlock();
+        }
+    }
 
+    public void startWatch(IContext ctx, TopoServer topoServer, String cell, Set<String> keyspaces) {
+        lock.lock();
+        try {
             if (!cellTopologyWatcherMap.containsKey(cell)) {
-                TopologyWatcher topologyWatcher = new TopologyWatcher(topoServer, cell, tabletKeyspace);
+                TopologyWatcher topologyWatcher = new TopologyWatcher(topoServer, cell, keyspaces);
                 topologyWatcher.start(ctx);
                 cellTopologyWatcherMap.put(cell, topologyWatcher);
             }
@@ -89,68 +74,14 @@ public void watch(IContext ctx, String cell, String tabletKeyspace) {
     }
 
     public void close() {
-        closeScheduledExecutor();
-
         for (Map.Entry<String, TopologyWatcher> entry : cellTopologyWatcherMap.entrySet()) {
             TopologyWatcher topologyWatcher = entry.getValue();
             topologyWatcher.close();
         }
         cellTopologyWatcherMap.clear();
-        globalKeyspacesMap.clear();
-    }
-
-    public void resetScheduledExecutor() {
-        closeScheduledExecutor();
-
-        scheduledExecutor = new ScheduledThreadPoolExecutor(1, new VtThreadFactoryBuilder.DefaultThreadFactory("reload-cell-schedule", true));
-        scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
-        scheduledExecutor.setRemoveOnCancelPolicy(true);
-    }
-
-    public void closeScheduledExecutor() {
-        scheduledExecutor.shutdownNow();
-        try {
-            int tryAgain = 3;
-            while (tryAgain > 0 && !scheduledExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
-                tryAgain--;
-            }
-        } catch (InterruptedException e) {
-            // We're shutting down anyway, so just ignore.
-        }
     }
 
     public boolean isWatching(String cell) {
         return cellTopologyWatcherMap.containsKey(cell);
     }
-
-    public void startTickerReloadCell(IContext globalContext, TopoServer topoServer, TimeUnit timeUnit) {
-        scheduledExecutor.scheduleWithFixedDelay(() -> {
-            try {
-                tickerUpdateCells(globalContext, topoServer);
-            } catch (Throwable e) {
-                LOGGER.error("tickerUpdateCells error: " + e);
-            }
-        }, 5, 10, timeUnit);
-    }
-
-    private void tickerUpdateCells(IContext globalContext, TopoServer topoServer) throws TopoException {
-        String serverAddress = topoServer.getServerAddress();
-        Set<String> keyspaceSet = globalKeyspacesMap.get(serverAddress);
-        if (CollectionUtils.isEmpty(keyspaceSet)) {
-            throw new RuntimeException("not found keyspace in " + serverAddress + " of TopologyWatcherManager.globalKeyspacesMap .");
-        }
-        List<String> allCells = topoServer.getAllCells(globalContext);
-        for (String cell : allCells) {
-            if (!isWatching(cell)) {
-                lock.lock();
-                try {
-                    TopologyWatcher topologyWatcher = new TopologyWatcher(topoServer, cell, keyspaceSet);
-                    topologyWatcher.start(globalContext);
-                    cellTopologyWatcherMap.put(cell, topologyWatcher);
-                } finally {
-                    lock.unlock();
-                }
-            }
-        }
-    }
 }

src/main/java/com/jd/jdbc/monitor/TopoServerCollector.java

@@ -0,0 +1,42 @@
+/*
+Copyright 2024 JD Project Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package com.jd.jdbc.monitor;
+
+import io.prometheus.client.Counter;
+
+public class TopoServerCollector {
+
+    private static final Counter EXEC_COUNTER = Counter.build()
+        .name("TopoServer_get_all_cell_counter")
+        .labelNames("Address")
+        .help("TopoServer get all cells counter")
+        .register(MonitorServer.getCollectorRegistry());
+
+    private static final Counter NEW_CELLS_COUNTER = Counter.build()
+        .name("TopoServer_new_cell_counter")
+        .labelNames("Address")
+        .help("TopoServer add new cell counter")
+        .register(MonitorServer.getCollectorRegistry());
+
+    public static Counter getExecCounterCounter() {
+        return EXEC_COUNTER;
+    }
+
+    public static Counter geCellsCounter() {
+        return NEW_CELLS_COUNTER;
+    }
+}

src/main/java/com/jd/jdbc/topo/Topo.java

@@ -18,6 +18,7 @@
 
 package com.jd.jdbc.topo;
 
+import com.jd.jdbc.context.IContext;
 import static com.jd.jdbc.topo.TopoExceptionCode.NO_IMPLEMENTATION;
 import static com.jd.jdbc.topo.TopoServer.CELLS_PATH;
 import static com.jd.jdbc.topo.TopoServer.CELL_INFO_FILE;
@@ -34,6 +35,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class Topo {
 
@@ -45,7 +47,7 @@ public class Topo {
 
     public static String TOPO_GLOBAL_ROOT = "/vitess/global";
 
-    public static TopoServer getTopoServer(TopoServerImplementType topoServerImplementType, String topoServerAddress) throws TopoException {
+    public static TopoServer getTopoServer(IContext ctx, TopoServerImplementType topoServerImplementType, String topoServerAddress) throws TopoException {
         synchronized (Topo.class) {
             registerFactory(topoServerImplementType);
 
@@ -58,6 +60,7 @@ public static TopoServer getTopoServer(TopoServerImplementType topoServerImpleme
                 throw new TopoException(Vtrpc.Code.UNKNOWN, "");
             }
             topoServers.put(topoServerAddress, topoServer);
+            topoServer.startTickerReloadCell(ctx);
             return topoServer;
         }
     }
@@ -124,7 +127,10 @@ protected static TopoServer newWithFactory(TopoFactory topoFactory, String serve
         topoServer.globalCell = conn;
         topoServer.globalReadOnlyCell = connReadOnly;
         topoServer.topoFactory = topoFactory;
-        topoServer.cells = new HashMap<>(16);
+        topoServer.cellsTopoConnMap = new HashMap<>(16);
+        topoServer.keyspaces = ConcurrentHashMap.newKeySet();
+        topoServer.cells = ConcurrentHashMap.newKeySet();
+        topoServer.localCell = "";
         topoServer.serverAddress = serverAddress;
         return topoServer;
     }