This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9da67831928 [fix](cloud) Drop CloudReplica route entries of dropped
compute groups (#66984)
9da67831928 is described below
commit 9da678319280d29bdaf2bc67888f780b887088ad
Author: deardeng <[email protected]>
AuthorDate: Tue Aug 25 17:05:25 2026 +0800
[fix](cloud) Drop CloudReplica route entries of dropped compute groups
(#66984)
`CloudReplica.primaryClusterToBackend` (`be`, previously `bes`) is
persisted per replica, but nothing ever removes an entry:
`CloudSystemInfoService .dropCluster()` only touches the system-info
maps, `CloudTabletRebalancer .loopCloudReplica()` only walks the compute
groups that currently exist, and `unprotectUpdateCloudReplica()` only
puts. So every created/dropped compute group leaves one key behind on
every replica, forever.
On an instance that had repeatedly added and removed compute groups this
grew to 392 route keys per replica against 3 live compute groups: 45.5M
entries over 167k replicas, 1.78 GB of the 2.07 GB image (85.9%) and
~8.6 GB of live heap. Metadata checkpoint keeps the online catalog and
the checkpoint catalog resident at the same time, which no longer fit in
the heap, so checkpoint OOMed on every leader in turn and the image
never advanced.
Drop an entry once its backend id no longer resolves. Such an entry is
never routed to: `getBackendIdImpl()` already resolves the id, gets null
and falls back to `hashReplicaToBe()`, so it only costs memory and image
bytes.
No edit log op is written for the removal, and none is needed:
`Checkpoint` is a `MasterDaemon`, and its Env cleans the catalog it
loads before saving, so the image is clean no matter what any serving
Env holds. The rebalancer sweep is master-only, so a follower keeps its
own stale entries in heap until it restarts or is promoted, at which
point one rebalancer round clears them. That is bounded, never reaches
persisted state, and is the tradeoff for not adding an edit log op that
would block downgrades.
A dead primary is kept while its compute group still has a live
secondary. With `enable_immediate_be_assign=false` -- the default --
primary pointing at an unavailable BE while the secondary holds the
rehashed live one is the normal failover state, and
`FrontendServiceImpl.getTabletReplicaInfos()` reaches secondaries
through `getPrimaryComputeGroupIds()`, so dropping the primary key would
hide a live secondary BE from the lazy fetch peer cache candidates. The
secondary map is cleaned first, so a dead pair is still fully removed.
Two call sites:
- `gsonPostProcess()`, outside the `bes` migration branch, so both
formats are covered. The backends module is loaded before db/recycleBin,
and the checkpoint thread resolves `Env.getCurrentEnv()` to its own Env,
so the backend set read there belongs to the image being loaded. This
shrinks the load peak and makes existing oversized images converge.
- The existing per-replica callback in `completeRouteInfo()`, so a
running leader converges too instead of waiting for a restart. It is
pinned to one compute group per round because `loopCloudReplica()`
invokes the callback once per (replica, compute group) while holding
`table.readLock()`.
`secondaryClusterToBackends` is not persisted, but leaks the same way in
heap, so it is cleaned by the same predicate.
Guarded by `enable_cloud_replica_stale_route_clean` (default true).
Measured on the metadata of the affected instance. Same build and same
`image.22568053` (2,096,733,909 B) for both runs, only the config
flipped; the
FE was started with `enable_check_compatibility_mode` and
`checkpoint_after_check_compatibility`, which loads the image, replays
the journal and dumps a new one.
| | clean=false | clean=true |
| --- | --- | --- |
| dumped image | 1,961,181,357 B | 272,087,919 B (-86.1%) |
| route entries | 46,004,776 | 353,422 (-99.2%) |
| distinct route keys | 395 | 2 |
| stored format | `bes` x 176,941 | `be` x 176,926 |
| live set after GC | 8.85 GB | 2.73 GB (-69%) |
| heap expanded to | 31 GB | stayed at the 8 GB Xms |
| dump duration | 72.6 s | 6.7 s |
The two surviving route keys are exactly the two compute groups whose
backends
are present in the image's own backend module -- `cluster_id_1`
-> 1760610536752 and `cluster_id_2` -> 1765176237562, both alive and
NORMAL -- so no live route was dropped.
Feeding the 272 MB result back in with `-Xmx8g` loads, replays and
re-dumps it
in 20s with a 0.51 GB live set and a stable output size, against the
19.2 GiB
heap that could not complete a checkpoint before.
---
.../main/java/org/apache/doris/common/Config.java | 7 +
.../apache/doris/cloud/catalog/CloudReplica.java | 76 ++++++++
.../doris/cloud/catalog/CloudTabletRebalancer.java | 62 ++++++-
.../doris/cloud/catalog/CloudReplicaTest.java | 206 +++++++++++++++++++++
.../cloud/catalog/CloudTabletRebalancerTest.java | 36 ++++
5 files changed, 386 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 b8c205cf041..c2998f89f34 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
@@ -3352,6 +3352,13 @@ public class Config extends ConfigBase {
+ "other BEs in cloud mode.")
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. If false, "
+ "use the legacy modulo placement. Restart-only.")
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 caec16887ac..b7ae1308ed5 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;
@@ -388,9 +389,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) {
@@ -402,6 +428,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;
@@ -596,6 +623,50 @@ 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.
+ // getBackendByIdWithBoxedId over getBackend: the ids here are already
boxed, and this loop runs
+ // per replica, so letting getBackend(long) unbox and rebox them would
allocate a Long per route.
+ int removed = 0;
+ if (!secondaryClusterToBackends.isEmpty()) {
+ for (Map.Entry<String, Pair<Long, Long>> entry :
secondaryClusterToBackends.entrySet()) {
+ if
(systemInfo.getBackendByIdWithBoxedId(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.getBackendByIdWithBoxedId(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.
@@ -669,5 +740,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 d4447e987fc..9e92587da5e 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;
@@ -1003,9 +1006,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;
@@ -1017,6 +1058,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
@@ -1085,7 +1136,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;
@@ -1808,6 +1860,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);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
index ca80fde8e0e..8b04976a490 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
@@ -24,8 +24,12 @@ 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.persist.gson.GsonUtils;
import org.apache.doris.system.Backend;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -433,4 +437,206 @@ public class CloudReplicaTest {
List<Backend> primaryBes = replica.getAllPrimaryBes();
Assertions.assertTrue(primaryBes.isEmpty());
}
+
+ // ---------------------------------------------------------------
+ private void stubBackend(long id, Backend be) {
+ Mockito.when(mockInfoService.getBackend(id)).thenReturn(be);
+
Mockito.when(mockInfoService.getBackendByIdWithBoxedId(id)).thenReturn(be);
+ }
+
+ // Tests for removeInvalidRoutes: routes of dropped compute groups
+ // ---------------------------------------------------------------
+
+ @Test
+ public void testRemoveInvalidRoutes() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ FeConstants.runningUnitTest = false;
+ Backend liveBe = createBackend(1001L, true, false);
+ stubBackend(1001L, liveBe);
+ // 2001 belongs to a compute group that has been dropped, so it no
longer resolves
+ stubBackend(2001L, null);
+
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 1001L);
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
+ replica.updateClusterToSecondaryBe(CLUSTER_ID_2, 2001L);
+
+ // one primary entry plus one secondary entry, both pointing at
the dropped backend
+ Assertions.assertEquals(2, replica.removeInvalidRoutes());
+
+ Assertions.assertEquals(1,
replica.getPrimaryComputeGroupIds().size());
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1));
+ // if the compute group id is reused later, the stale entry must
not come back to life
+ Backend revivedBe = createBackend(2001L, true, false);
+ stubBackend(2001L, revivedBe);
+ Assertions.assertNull(replica.getSecondaryBackend(CLUSTER_ID_2));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testRemoveInvalidRoutes_disabled() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = false;
+ FeConstants.runningUnitTest = false;
+ stubBackend(2001L, null);
+
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
+ replica.updateClusterToSecondaryBe(CLUSTER_ID_2, 2001L);
+
+ Assertions.assertEquals(0, replica.removeInvalidRoutes());
+
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_2));
+ Backend revivedBe = createBackend(2001L, true, false);
+ stubBackend(2001L, revivedBe);
+
Assertions.assertNotNull(replica.getSecondaryBackend(CLUSTER_ID_2));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testRemoveInvalidRoutes_keepsDeadPrimaryWithLiveSecondary() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ FeConstants.runningUnitTest = false;
+ Backend liveBe = createBackend(1001L, true, false);
+ stubBackend(1001L, liveBe);
+ stubBackend(2001L, null);
+
+ // the normal failover state with
enable_immediate_be_assign=false: primary still points at the
+ // old backend while the secondary holds the rehashed live one
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 2001L);
+ replica.updateClusterToSecondaryBe(CLUSTER_ID_1, 1001L);
+
+ Assertions.assertEquals(0, replica.removeInvalidRoutes());
+
+ // FrontendServiceImpl's lazy fetch reaches the secondary through
the primary key set
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1));
+
Assertions.assertNotNull(replica.getSecondaryBackend(CLUSTER_ID_1));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ private CloudReplica gsonRoundTrip(CloudReplica replica, boolean
legacyBesFormat) {
+ JsonObject json = GsonUtils.GSON.toJsonTree(replica,
Replica.class).getAsJsonObject();
+ if (legacyBesFormat) {
+ // rewrite the scalar `be` map back into the pre-#59932 `bes` map
of single element lists
+ JsonObject be = json.remove("be").getAsJsonObject();
+ JsonObject bes = new JsonObject();
+ for (String cg : be.keySet()) {
+ JsonArray beIds = new JsonArray();
+ beIds.add(be.get(cg).getAsLong());
+ bes.add(cg, beIds);
+ }
+ json.add("bes", bes);
+ }
+ return (CloudReplica) GsonUtils.GSON.fromJson(json, Replica.class);
+ }
+
+ @Test
+ public void testRemoveInvalidRoutes_onImageLoad() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ FeConstants.runningUnitTest = false;
+ Backend liveBe = createBackend(1001L, true, false);
+ stubBackend(1001L, liveBe);
+ stubBackend(2001L, null);
+
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 1001L);
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
+
+ // both the current `be` format and the legacy `bes` format must
be cleaned on load,
+ // so the cleanup has to sit outside the bes -> be migration
branch of gsonPostProcess()
+ for (boolean legacy : new boolean[] {false, true}) {
+ CloudReplica loaded = gsonRoundTrip(replica, legacy);
+ Assertions.assertEquals(1,
loaded.getPrimaryComputeGroupIds().size(), "legacy=" + legacy);
+
Assertions.assertTrue(loaded.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1),
"legacy=" + legacy);
+ }
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testRouteDiscardedWhenBackendVanishesDuringAssign() throws
ComputeGroupException {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ boolean savedImmediate = Config.enable_immediate_be_assign;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ Config.enable_immediate_be_assign = true;
+ FeConstants.runningUnitTest = false;
+
Mockito.when(mockColocateIndex.isColocateTableNoLock(TABLE_ID)).thenReturn(false);
+
+ // the compute group still lists be 1001, so hashReplicaToBe()
picks it ...
+ Backend be = createBackend(1001L, true, false);
+ Mockito.when(mockInfoService.getBackendsByClusterId(CLUSTER_ID_1))
+ .thenReturn(new ArrayList<>(Arrays.asList(be)));
+
Mockito.when(mockInfoService.getClusterNameByClusterId(CLUSTER_ID_1)).thenReturn(CLUSTER_NAME_1);
+ // ... but by the time the route is published the group has been
dropped and 1001 is gone
+ stubBackend(1001L, null);
+
+ CloudReplica replica = createReplica();
+ replica.getBackendIdWithClusterId(CLUSTER_ID_1);
+
+ // the write must not survive: nothing else would ever clean it,
the sweep already ran
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().isEmpty());
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ Config.enable_immediate_be_assign = savedImmediate;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testDiscardRouteReportsWhetherTheRouteSurvived() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ FeConstants.runningUnitTest = false;
+ Backend liveBe = createBackend(1001L, true, false);
+ stubBackend(1001L, liveBe);
+ stubBackend(2001L, null);
+
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 1001L);
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
+
+ // backend still registered: the drop can only come later, so the
sweep it triggers covers us
+
Assertions.assertTrue(replica.discardRouteIfBackendVanished(1001L));
+ // backend already gone: the caller must not persist this route,
and it is dropped here
+
Assertions.assertFalse(replica.discardRouteIfBackendVanished(2001L));
+
Assertions.assertFalse(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_2));
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1));
+
+ // with the switch off the legacy behaviour is kept: nothing is
discarded or reported
+ Config.enable_cloud_replica_stale_route_clean = false;
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
+
Assertions.assertTrue(replica.discardRouteIfBackendVanished(2001L));
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_2));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
index be064ee442a..f2b630947f1 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudTabletRebalancerTest.java
@@ -49,6 +49,7 @@ import java.lang.reflect.Modifier;
import java.util.AbstractList;
import java.util.AbstractMap;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -1224,4 +1225,39 @@ public class CloudTabletRebalancerTest {
"compute_cluster_b", "cluster-b",
CloudTabletRebalancer.StatType.PARTITION, 0L));
}
}
+
+ @Test
+ public void testStaleRouteSweepGate() {
+ boolean saved = Config.enable_cloud_replica_stale_route_clean;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ TestRebalancer rebalancer = new TestRebalancer();
+ Set<Long> bes = new HashSet<>(Arrays.asList(1L, 2L));
+
+ // no baseline yet, so sweep -- twice, to catch route writers in
flight during the first pass
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes));
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes));
+ // unchanged topology: nothing can have gone stale
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(bes));
+
+ // a new backend strands nothing, so it does not trigger a sweep
on its own
+ Set<Long> grown = new HashSet<>(Arrays.asList(1L, 2L, 3L));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(grown));
+ // ... but it must have entered the baseline, so dropping it again
is still seen as a removal
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L, 2L))));
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L, 2L))));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L, 2L))));
+
+ // a backend that disappeared triggers a sweep
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L))));
+
+ // turning the switch off drops the baseline so turning it back on
sweeps again
+ Config.enable_cloud_replica_stale_route_clean = false;
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(bes));
+ Config.enable_cloud_replica_stale_route_clean = true;
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = saved;
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
