This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 34b443418ae [fix](cloud) Drop CloudReplica route entries of dropped
compute groups (#66984) (#67234)
34b443418ae is described below
commit 34b443418ae536849a9bd86a09e36f346060f98b
Author: deardeng <[email protected]>
AuthorDate: Mon Aug 31 16:23:16 2026 +0800
[fix](cloud) Drop CloudReplica route entries of dropped compute groups
(#66984) (#67234)
pick from https://github.com/apache/doris/pull/66984
Remove stale CloudReplica routes when compute-group backends are
dropped.
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../main/java/org/apache/doris/common/Config.java | 7 ++
.../apache/doris/cloud/catalog/CloudReplica.java | 74 ++++++++++++++++++++++
.../doris/cloud/catalog/CloudTabletRebalancer.java | 62 +++++++++++++++++-
3 files changed, 142 insertions(+), 1 deletion(-)
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 126b29998eb..814a10024fb 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 class Config extends ConfigBase {
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 40d55aa2255..4ad4f24dc7c 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.catalog.Replica;
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;
@@ -389,9 +390,34 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
} 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) {
@@ -403,6 +429,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
try {
beId = getBackendIdImpl(clusterId);
updateClusterToPrimaryBe(clusterId, beId);
+ discardRouteIfBackendVanished(beId);
return Env.getCurrentSystemInfo().getBackend(beId);
} catch (ComputeGroupException e) {
return null;
@@ -597,6 +624,48 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
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<String, Pair<Long, Long>> 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<String, Long> 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.
@@ -670,5 +739,10 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
}
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 11989525c4f..1434577ce32 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
@@ -105,6 +105,9 @@ public class CloudTabletRebalancer extends MasterDaemon {
private Map<String, List<Long>> clusterToBes;
private Set<Long> allBes;
+ // backend baseline and remaining sweep rounds, see staleRouteSweepNeeded()
+ private Set<Long> lastSweptBackends = null;
+ private int pendingSweepRounds = 0;
// partitionId -> indexId -> be -> tabletIds
private ConcurrentHashMap<Long, ConcurrentHashMap<Long,
ConcurrentHashMap<Long, Set<Long>>>> partitionToTablets;
@@ -966,9 +969,47 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
}
+ /**
+ * 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<Long> 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<UpdateCloudReplicaInfo> updateReplicaInfos = new
ArrayList<UpdateCloudReplicaInfo>();
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;
@@ -980,6 +1021,16 @@ public class CloudTabletRebalancer extends MasterDaemon {
for (Tablet tablet : tablets) {
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);
// colocate table no need to update primary backends
@@ -1048,7 +1099,8 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
});
- 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;
@@ -1773,6 +1825,14 @@ public class CloudTabletRebalancer extends MasterDaemon {
}
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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
