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;
  }