diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 126b29998eba5f..814a10024fbdf2 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -3807,6 +3807,13 @@ public static int metaServiceRpcRetryTimes() { description = { "存算分离模式下，一个 BE 挂掉多长时间后，它的 tablet 彻底转移到其他 BE 上" }) public static int rehash_tablet_after_be_dead_seconds = 3600; + @ConfField(mutable = true, masterOnly = false, + description = "Whether to drop the primary/secondary route entries of a CloudReplica whose backend no " + + "longer exists, when loading the image and in the tablet rebalancer round. Those entries are " + + "already ignored at query time (the replica is rehashed), so they only waste FE memory and " + + "image size. Set to false to keep the legacy leaking behavior. Default is true.") + public static boolean enable_cloud_replica_stale_route_clean = true; + @ConfField(mutable = false, masterOnly = true, description = { "Whether to use rendezvous hashing for colocate bucket placement in cloud mode. " diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java index 347416d972c3b8..a6d4bb81dc0101 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java @@ -23,6 +23,7 @@ import org.apache.doris.cloud.qe.ComputeGroupException; import org.apache.doris.cloud.system.CloudSystemInfoService; import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; import org.apache.doris.common.util.DebugPointUtil; import org.apache.doris.persist.gson.GsonPostProcessable; @@ -385,9 +386,34 @@ clusterId, pickBeId, Config.enable_immediate_be_assign, this, getPrimaryBackend( } else { updateClusterToSecondaryBe(clusterId, pickBeId); } + discardRouteIfBackendVanished(pickBeId); return pickBeId; } + /** + * The compute group can be dropped while a publisher is picking a backend out of it, in which case the + * route just written would outlive the group. Re-checking right after publishing closes that window: + * either the backend is already gone and the write is undone here, or it was still registered, which + * means the drop -- and the rebalancer sweep it triggers -- happens after this write and will visit + * this replica. Ordering, not timing, is what makes the sweep sufficient; a publisher may stall for + * arbitrarily long between choosing a backend and publishing without escaping cleanup. + * + * Every route publisher that resolves a backend outside the rebalancer round must call this, and must + * not persist the route when it returns false. + * + * @return false when the route was discarded because its backend is gone + */ + public boolean discardRouteIfBackendVanished(long beId) { + if (!Config.enable_cloud_replica_stale_route_clean) { + return true; + } + if (Env.getCurrentSystemInfo().getBackend(beId) == null) { + removeInvalidRoutes(); + return false; + } + return true; + } + public Backend getPrimaryBackend(String clusterId, boolean setIfAbsent) { long beId = getClusterPrimaryBackendId(clusterId); if (beId != -1L) { @@ -399,6 +425,7 @@ public Backend getPrimaryBackend(String clusterId, boolean setIfAbsent) { try { beId = getBackendIdImpl(clusterId); updateClusterToPrimaryBe(clusterId, beId); + discardRouteIfBackendVanished(beId); return Env.getCurrentSystemInfo().getBackend(beId); } catch (ComputeGroupException e) { return null; @@ -593,6 +620,48 @@ public void clearClusterToBe(String cluster) { secondaryClusterToBackends.remove(cluster); } + /** + * Drop the route entries whose backend has been dropped from the cluster. + * + * Such an entry is already dead weight: getBackendIdImpl() resolves the backend id, gets null and + * falls back to hashReplicaToBe(), so removing it does not change routing. But nothing ever removes + * it either -- dropCluster() only touches CloudSystemInfoService, and the rebalancer only walks the + * compute groups that currently exist -- so entries of dropped compute groups pile up forever, both + * in FE heap and in the image (the `bes`/`be` field). + * + * @return how many entries were dropped + */ + public int removeInvalidRoutes() { + if (!Config.enable_cloud_replica_stale_route_clean || FeConstants.runningUnitTest) { + return 0; + } + SystemInfoService systemInfo = Env.getCurrentSystemInfo(); + // Remove conditionally rather than by predicate: route writers run concurrently, so comparing map + // sizes before and after would mix their insertions into the count (and could even report a + // negative one), and a key whose value was just rewritten to a live backend must not be dropped. + int removed = 0; + if (!secondaryClusterToBackends.isEmpty()) { + for (Map.Entry> entry : secondaryClusterToBackends.entrySet()) { + if (systemInfo.getBackend(entry.getValue().key()) == null + && secondaryClusterToBackends.remove(entry.getKey(), entry.getValue())) { + removed++; + } + } + } + // Keep a dead primary whose compute group still has a live secondary. With + // enable_immediate_be_assign=false that is the normal failover state, and the lazy fetch path in + // FrontendServiceImpl.getTabletReplicaInfos() enumerates secondaries through the primary key set, + // so dropping the key would hide a live secondary from the peer cache candidates. + for (Map.Entry entry : primaryClusterToBackend.entrySet()) { + if (systemInfo.getBackend(entry.getValue()) == null + && !secondaryClusterToBackends.containsKey(entry.getKey()) + && primaryClusterToBackend.remove(entry.getKey(), entry.getValue())) { + removed++; + } + } + return removed; + } + /** * Returns the set of compute group IDs that have primary backends for this replica. * Used by lazy fetch path to also collect secondary backends per compute group. @@ -666,5 +735,10 @@ public void gsonPostProcess() throws IOException { } this.primaryClusterToBackends = null; } + // outside the `bes` branch on purpose: the new `be` format accumulates stale entries just the same. + // The backends module is loaded before db/recycleBin (PersistMetaModules.MODULE_NAMES), and the + // checkpoint thread resolves Env.getCurrentEnv() to its own Env, so the backend set read here is + // the one belonging to the image being loaded. + removeInvalidRoutes(); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java index 82d29f3d701f68..e0f0c6fc288745 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java @@ -52,6 +52,7 @@ import org.apache.doris.thrift.TWarmUpCacheAsyncRequest; import org.apache.doris.thrift.TWarmUpCacheAsyncResponse; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.base.Strings; import com.google.common.collect.Sets; @@ -97,6 +98,9 @@ public class CloudTabletRebalancer extends MasterDaemon { private Map> clusterToBes; private Set allBes; + // backend baseline and remaining sweep rounds, see staleRouteSweepNeeded() + private Set lastSweptBackends = null; + private int pendingSweepRounds = 0; // partitionId -> indexId -> be -> tabletIds private ConcurrentHashMap>>> partitionToTablets; @@ -933,9 +937,47 @@ public void checkDecommissionState(Map> clusterToBes) { } } + /** + * Decides whether this round sweeps stale routes, and advances the backend baseline. Call once per + * round. Without this gate the sweep would walk every replica's route maps once a second under + * table.readLock() only to find nothing, which on a large catalog is pure allocation. + */ + @VisibleForTesting + boolean staleRouteSweepNeeded(Set currentBes) { + if (!Config.enable_cloud_replica_stale_route_clean) { + lastSweptBackends = null; + pendingSweepRounds = 0; + return false; + } + if (lastSweptBackends == null || !currentBes.containsAll(lastSweptBackends)) { + // Only a backend that went away can strand a route. Two rounds rather than one: a query + // thread can pick a backend in hashReplicaToBe() before the drop and publish the route in + // getBackendIdImpl() after this pass already visited that replica, so the extra round + // catches writers that were in flight during the first one. + pendingSweepRounds = 2; + } + // An addition strands nothing, so it does not trigger a sweep -- but it must still enter the + // baseline. Advancing only after a sweep would leave the baseline at the pre-addition set, and + // dropping that same backend later would compare equal to it and go unnoticed. + lastSweptBackends = currentBes; + if (pendingSweepRounds > 0) { + pendingSweepRounds--; + return true; + } + return false; + } + private boolean completeRouteInfo() { List updateReplicaInfos = new ArrayList(); long[] assignedErrNum = {0L}; + long[] staleRouteNum = {0L}; + boolean sweepStaleRoutes = staleRouteSweepNeeded(allBes); + // loopCloudReplica() has the compute group loop innermost, so it hands us every replica once per + // live compute group, while removeInvalidRoutes() scans the whole route map and does not care + // which group we are on. Pin the sweep to one arbitrary group id so a sweeping round still makes a + // single pass per replica. If clusterToBes is empty the callback never runs at all, so the serving + // catalog keeps the entries until it reloads the image -- there is nothing to route in that state. + String sweepTicket = sweepStaleRoutes ? clusterToBes.keySet().stream().findFirst().orElse(null) : null; long needRehashDeadTime = System.currentTimeMillis() - Config.rehash_tablet_after_be_dead_seconds * 1000L; loopCloudReplica((Database db, Table table, Partition partition, MaterializedIndex index, String cluster) -> { boolean assigned = false; @@ -945,6 +987,16 @@ private boolean completeRouteInfo() { for (Tablet tablet : index.getTablets()) { for (Replica r : tablet.getReplicas()) { CloudReplica replica = (CloudReplica) r; + // Drop routes of compute groups that no longer exist; gsonPostProcess() only converges + // the catalog on image load, so without this the leader keeps them until it restarts. + // No edit log op is written for the removal: the entries are already unroutable, and + // the image is written by the master-only checkpoint, whose Env cleans the catalog it + // loads, so no leader/follower difference ever reaches persisted state. Note this + // daemon is master-only, so a follower keeps its own stale entries in heap until it + // restarts or is promoted and runs a round here. + if (cluster.equals(sweepTicket)) { + staleRouteNum[0] += replica.removeInvalidRoutes(); + } // clean secondary map replica.checkAndClearSecondaryClusterToBe(cluster, needRehashDeadTime); InfightTablet taskKey = new InfightTablet(tablet.getId(), cluster); @@ -1011,7 +1063,8 @@ private boolean completeRouteInfo() { } }); - LOG.info("collect to editlog route {} infos, error num {}", updateReplicaInfos.size(), assignedErrNum[0]); + LOG.info("collect to editlog route {} infos, error num {}, swept stale routes {}, entries dropped {}", + updateReplicaInfos.size(), assignedErrNum[0], sweepStaleRoutes, staleRouteNum[0]); if (updateReplicaInfos.isEmpty()) { return true; @@ -1678,6 +1731,14 @@ private void updateClusterToBeMap(long tabletId, long destBe, String clusterId, } cloudReplica.updateClusterToPrimaryBe(clusterId, destBe); + if (!cloudReplica.discardRouteIfBackendVanished(destBe)) { + // The warmup checker resolves its destination on its own scheduler thread and can stall + // there while the compute group is dropped and the sweeps triggered by that drop finish. + // Journalling the route now would hand every FE an entry nothing is left to clean. + LOG.info("compute group {} lost backend {} while warming up tablet {}, dropping the route", + clusterId, destBe, tabletId); + return; + } UpdateCloudReplicaInfo info = new UpdateCloudReplicaInfo(tabletMeta.getDbId(), tabletMeta.getTableId(), tabletMeta.getPartitionId(), tabletMeta.getIndexId(), tabletId, cloudReplica.getId(), clusterId, destBe);