This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 ed735be66ef [fix](cloud) Clean stale routes when replaying replica
updates and before saving checkpoint images (#67636)
ed735be66ef is described below
commit ed735be66efbef4c94abc9a2498321a22237a07d
Author: deardeng <[email protected]>
AuthorDate: Wed Oct 14 15:16:44 2026 +0800
[fix](cloud) Clean stale routes when replaying replica updates and before
saving checkpoint images (#67636)
Problem Summary:
A CloudReplica keeps one route entry per compute group. When a compute
group is dropped, the master sweeps the stale entries out of its own
heap in CloudTabletRebalancer, but that sweep writes no edit log. A
follower only replays OP_UPDATE_CLOUD_REPLICA, which adds or overwrites
the route of the compute group named in the log and never removes
anything, so on a follower the route maps only ever grow. With a fixed
set of replicas and repeated compute group create/drop rounds, the
follower's route entries grow linearly with the number of rounds while
the master stays flat.
The generated checkpoint image has a related but separate gap. The
checkpoint Env does loadImage -> replayJournal -> post-processing ->
saveImage. gsonPostProcess() cleans what loadImage deserializes, but
nothing cleans what the replay window then adds, and
postProcessCloudMetadata() only copies tablet stats. So every image is
written with the stale routes accumulated since the previous image.
An earlier version of this PR fixed both by calling
CloudReplica.removeInvalidRoutes() once per replayed
UpdateCloudReplicaInfo journal entry, plus once before saving a
checkpoint image. That works, but on a large catalog (tens of millions
of tablets, many compute groups) it means every single route-update
journal entry re-scans that replica's whole route map, which adds up
across a replay batch and slows down replay itself -- the thing route
cleanup should never be on the critical path for.
This revision keeps the checkpoint-time sweep but replaces the
per-journal-entry replay cleanup with a coalesced sweep over the active
catalog, run at most once per replay batch, and adds an equivalent sweep
to the master's rebalancer round (previously that sweep was piggybacked
on the rebalancer's per-compute-group replica loop and had no way to run
when the cluster's live compute-group set had just gone empty). It also
fixes a correctness gap in the original per-journal-entry design: a
journal written before a backend was dropped can be replayed after the
drop, and the original code published that route unconditionally before
the next cleanup pass could catch it, which could momentarily discard an
already-valid secondary route for the same compute group.
Three independent sweep triggers, all reusing
CloudInternalCatalog.removeInvalidCloudReplicaRoutes(SystemInfoService)
-- an instance method that walks this catalog's own
db/table/partition/index(ALL, including shadow indexes)/tablet/replica
tree and calls CloudReplica.removeInvalidRoutes(systemInfo) per replica:
1. CloudEnv.replayJournal() calls it once per replay batch, gated on
CloudSystemInfoService.replayBackendRemovalVersion having changed since
the last sweep (a counter bumped only by replayDropBackend()). Additions
alone never strand a route, so they don't trigger a sweep. The very
first replay after startup always sweeps once regardless of the counter.
A failed sweep is retried on the next batch rather than being silently
dropped. The checkpoint thread's own replayJournal() call skips this
path entirely (Checkpoint.doCheckpoint() sweeps explicitly after replay
finishes, see 3 below).
2. CloudTabletRebalancer.completeRouteInfo() sweeps once per round when
the live backend set has shrunk since the last sweep, independent of the
per-compute-group replica loop (the old code piggybacked the sweep on
that loop by pinning it to one arbitrary compute group, which meant a
topology with no live compute groups could never trigger a sweep at
all). It runs for two consecutive rounds after a drop is detected, not
one: a query thread can pick a backend in hashReplicaToBe() just before
the drop and publish the route in getBackendIdImpl() after this round's
pass has already visited that replica, so the extra round catches routes
published while the first one was still in flight. This sweep can now
also be confined to a configurable daily time window
(cloud_tablet_rebalancer_stale_route_clean_start_time /
cloud_tablet_rebalancer_stale_route_clean_end_time, HH:mm, both default
to "00:00" which is unrestricted); outside the window the
pending-sweep-round counter is left untouched rather than drained, so a
drop detected outside the window still gets its two rounds once the
window opens.
3. Checkpoint.doCheckpoint() still sweeps the checkpoint Env's own
catalog against its own backend set after replay finishes and before
saveImage(), now via an explicit SystemInfoService parameter
(env.getClusterInfo()) threaded down through the whole call chain
instead of being resolved internally via
Env.getCurrentSystemInfo()/Env.getCurrentEnv(). This removes the
implicit dependency on the checkpoint thread's identity to resolve the
right Env, and keeps the catalog being swept and the backend set
deciding staleness from ever being able to come from two different Envs.
Late-journal protection:
CloudReplica.replayUpdateClusterToPrimaryBe(clusterId, beId, systemInfo)
replaces the old unconditional updateClusterToPrimaryBe() +
removeInvalidRoutes() pair in
CloudInternalCatalog.unprotectUpdateCloudReplica(). Before publishing a
replayed primary-route update, it checks whether the target backend is
still registered; if not, the update is dropped instead of being
published and then possibly torn down by the next sweep, which
previously had a window where it could also clear an already-valid
secondary route for the same compute group before the primary write was
reconciled.
FrontendServiceImpl.getTabletReplicaInfos()'s lazy peer-discovery path
now collects secondary backends via CloudReplica.getAllSecondaryBes()
(which scans secondaryClusterToBackends directly) instead of only
looking up a secondary for compute groups that still have a primary key
present. A secondary-only route (primary key already swept, secondary
still valid) was previously invisible to this path.
The journal protocol is unchanged: no new opcode, no new field, no
delete event, and no change to the meaning of existing logs.
Existing cleanup semantics are reused as-is:
enable_cloud_replica_stale_route_clean still gates every sweep call
site, staleness is still decided by whether the backend is registered in
the relevant Env (not by isAlive() or heartbeat), a dead primary whose
compute group still has a live secondary is still kept for the failover
and peer cache paths, and removal is still the conditional remove(key,
expectedValue).
Boundaries, deliberately not addressed here:
- Cleanup remains a synchronous full-catalog scan at each trigger point;
budgeted/incremental scheduling is deferred to a follow-up PR and should
be based on measured replay latency at production tablet-count scale,
not assumed here.
- The rebalancer's sweep only runs on the master; on a Follower, only
the replay-batch sweep and image-load's gsonPostProcess() reclaim
routes. That asymmetry is unchanged from before this PR.
- CatalogRecycleBin is intentionally out of scope: this PR only sweeps
the active catalog. A recycled table or partition's routes are reclaimed
through the normal recycle-bin lifecycle (erase, or restored and picked
up by a later active-catalog sweep after RECOVER), not by this sweep.
- A backend id that still exists but now belongs to a different compute
group is not detected, as before.
### Test
- run-fe-ut.sh across CloudEnvRouteCleanupTest, CloudReplicaTest,
CloudTabletRebalancerTest, CloudInternalCatalogTest, CheckpointTest,
FrontendServiceImplCloudTest, CatalogRecycleBinTest and RecoverTest: 106
tests, 0 failures, 0 errors. FE build with checkstyle passes with 0
violations (fe-common and fe-core).
- CloudEnvRouteCleanupTest (new): coalescing multiple backend-removal
journals into one sweep per batch, sweeping once on startup and once
immediately after re-enabling the config, retrying a sweep that threw,
and skipping the sweep entirely on the checkpoint thread.
- CloudInternalCatalogTest: replaying a route update no longer scans or
removes other routes itself (that responsibility moved to the batch
sweep); a late journal to an already-dropped backend does not clear an
existing valid secondary route.
- CheckpointTest: stale route dropped while a live route is kept; a dead
primary with a live secondary is kept; shadow-index and
temporary-partition replicas are swept; a recycled table/partition's
replicas are not swept and CatalogRecycleBin is never touched; cleanup
disabled is a no-op; a new case asserts the sweep judges backend
liveness against the checkpoint Env's own backend set, not the serving
Env's, by giving the two a different opinion on the same backend id and
checking which one wins.
- CloudTabletRebalancerTest: the sweep now runs independent of the
per-compute-group loop, including when the live compute-group set is
empty; the two-round-after-a-drop behavior; the new HH:mm time window,
including the midnight-wrapping case and that pending sweep rounds are
preserved (not drained) while outside the window.
- FrontendServiceImplCloudTest (new case): a secondary-only route (no
primary key present for that compute group) is still returned by the
lazy peer-discovery path.
- Not covered by tests: no multi-FE cluster regression, and no
large-scale (tens-of-millions-of-tablets) performance run reproducing
the original replay-latency numbers.
### Release note
Coalesce stale CloudReplica route cleanup to run once per replay batch
(and once per rebalancer round, now with an optional daily time window)
instead of on every route-update journal entry, and stop a late journal
from clearing an already-valid secondary route.
---
.../main/java/org/apache/doris/common/Config.java | 21 +-
.../org/apache/doris/cloud/catalog/CloudEnv.java | 27 ++
.../apache/doris/cloud/catalog/CloudReplica.java | 58 ++++-
.../doris/cloud/catalog/CloudTabletRebalancer.java | 72 ++++--
.../cloud/datasource/CloudInternalCatalog.java | 92 ++++++-
.../doris/cloud/system/CloudSystemInfoService.java | 8 +
.../java/org/apache/doris/master/Checkpoint.java | 19 ++
.../apache/doris/service/FrontendServiceImpl.java | 7 +-
.../cloud/catalog/CloudEnvRouteCleanupTest.java | 145 +++++++++++
.../doris/cloud/catalog/CloudReplicaTest.java | 6 +-
.../cloud/catalog/CloudTabletRebalancerTest.java | 119 ++++++++-
.../cloud/datasource/CloudInternalCatalogTest.java | 184 ++++++++++++++
.../org/apache/doris/master/CheckpointTest.java | 276 +++++++++++++++++++++
.../service/FrontendServiceImplCloudTest.java | 35 +++
14 files changed, 1019 insertions(+), 50 deletions(-)
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 487efc4114d..076723aa8ac 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
@@ -3571,11 +3571,26 @@ public class Config extends ConfigBase {
@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.")
+ + "longer exists, when loading the image, in the tablet
rebalancer round, and once per replayed "
+ + "batch of backend-removal journals (on Followers too,
since each FE applies this locally). "
+ + "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 = true, masterOnly = true, description = "Start time of
the window in which the cloud "
+ + "tablet rebalancer is allowed to sweep stale CloudReplica
routes, in HH:mm format. Used with "
+ + "`cloud_tablet_rebalancer_stale_route_clean_end_time` to decide
the window. Only gates the "
+ + "rebalancer's own sweep -- image loading and replayed
backend-removal batches always clean "
+ + "regardless of this window. If set to the same value as the end
time (the default), the "
+ + "rebalancer sweeps any time it would otherwise sweep, i.e. the
window is unrestricted.")
+ public static String cloud_tablet_rebalancer_stale_route_clean_start_time
= "00:00";
+
+ @ConfField(mutable = true, masterOnly = true, description = "End time of
the window in which the cloud "
+ + "tablet rebalancer is allowed to sweep stale CloudReplica
routes, in HH:mm format. See "
+ + "`cloud_tablet_rebalancer_stale_route_clean_start_time`.")
+ public static String cloud_tablet_rebalancer_stale_route_clean_end_time =
"00:00";
+
@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/CloudEnv.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
index 85103b73487..681283576a6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnv.java
@@ -94,6 +94,33 @@ public class CloudEnv extends Env {
private String clusterSnapshotFile;
+ private boolean replayRoutesCleaned;
+ private long cleanedBackendRemovalVersion;
+
+ @Override
+ public synchronized boolean replayJournal(long toJournalId) {
+ boolean hasLog = super.replayJournal(toJournalId);
+ // Checkpoint sweeps after postProcessCloudMetadata(), immediately
before saving its image.
+ if (Env.isCheckpointThread()) {
+ return hasLog;
+ }
+ if (!Config.enable_cloud_replica_stale_route_clean) {
+ replayRoutesCleaned = false;
+ return hasLog;
+ }
+ CloudSystemInfoService systemInfo = (CloudSystemInfoService)
getClusterInfo();
+ long removalVersion = systemInfo.getReplayBackendRemovalVersion();
+ if (!replayRoutesCleaned || removalVersion !=
cleanedBackendRemovalVersion) {
+ // ponytail: one synchronous scan per affected batch; budgeted
cleanup belongs in a follow-up.
+ // Pass this CloudEnv's own catalog and backend set together so
the sweep can never judge one
+ // Env's replicas against another Env's backends, regardless of
which thread calls replayJournal.
+ ((CloudInternalCatalog)
getInternalCatalog()).removeInvalidCloudReplicaRoutes(systemInfo);
+ cleanedBackendRemovalVersion = removalVersion;
+ replayRoutesCleaned = true;
+ }
+ return hasLog;
+ }
+
public CloudEnv(boolean isCheckpointCatalog) {
super(isCheckpointCatalog);
this.cleanCopyJobScheduler = new CleanCopyJobScheduler();
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 3d6628985c8..a80df69a579 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
@@ -416,8 +416,11 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
if (!Config.enable_cloud_replica_stale_route_clean) {
return true;
}
- if (Env.getCurrentSystemInfo().getBackend(beId) == null) {
- removeInvalidRoutes();
+ // Query-path self-check, not part of the sweep/replay chain: this
always runs on whatever thread
+ // is serving the request, never the checkpoint thread, so
Env.getCurrentSystemInfo() is correct here.
+ SystemInfoService systemInfo = Env.getCurrentSystemInfo();
+ if (systemInfo.getBackend(beId) == null) {
+ removeInvalidRoutes(systemInfo);
return false;
}
return true;
@@ -646,13 +649,17 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
* 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).
*
+ * @param systemInfo the backend set to judge staleness against. Callers
must pass the SystemInfoService
+ * of the same Env whose catalog this replica belongs
to (the serving Env during normal
+ * replay/rebalancing, or the checkpoint's private Env
during image generation) -- never
+ * resolve it internally via
Env.getCurrentSystemInfo(), since that depends on which
+ * thread happens to be calling and would silently
judge the wrong Env's replicas.
* @return how many entries were dropped
*/
- public int removeInvalidRoutes() {
+ public int removeInvalidRoutes(SystemInfoService systemInfo) {
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.
@@ -664,18 +671,24 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
if
(systemInfo.getBackendByIdWithBoxedId(entry.getValue().key()) == null
&& secondaryClusterToBackends.remove(entry.getKey(),
entry.getValue())) {
removed++;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("removed stale secondary route, clusterId
{}, beId {}, replica info {}",
+ entry.getKey(), entry.getValue().key(), this);
+ }
}
}
}
- // 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.
+ // Preserve the primary marker for the normal failover state with a
live secondary.
+ // Peer discovery enumerates secondaries independently: publication
can race with this check.
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++;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("removed stale primary route, clusterId {}, beId
{}, replica info {}",
+ entry.getKey(), entry.getValue(), this);
+ }
}
}
return removed;
@@ -683,7 +696,6 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
/**
* 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.
*/
public Set<String> getPrimaryComputeGroupIds() {
return primaryClusterToBackend.keySet();
@@ -708,6 +720,32 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
}
}
+ public List<Backend> getAllSecondaryBes() {
+ List<Backend> result = new ArrayList<>();
+ SystemInfoService systemInfo = Env.getCurrentSystemInfo();
+ for (Pair<Long, Long> route : secondaryClusterToBackends.values()) {
+ Backend backend =
systemInfo.getBackendByIdWithBoxedId(route.first);
+ if (backend != null) {
+ result.add(backend);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * @param systemInfo the backend set to judge staleness against -- see
{@link #removeInvalidRoutes} for why
+ * this must come from the caller's Env rather than
Env.getCurrentSystemInfo().
+ */
+ public void replayUpdateClusterToPrimaryBe(String clusterId, long beId,
SystemInfoService systemInfo) {
+ // A journal queued before backend deletion can arrive after the
deletion was replayed.
+ // Do not publish that route, or clear a live secondary while
rejecting it.
+ if (Config.enable_cloud_replica_stale_route_clean &&
!FeConstants.runningUnitTest
+ && systemInfo.getBackend(beId) == null) {
+ return;
+ }
+ updateClusterToPrimaryBe(clusterId, beId);
+ }
+
public List<Backend> getAllPrimaryBes() {
List<Backend> result = new ArrayList<Backend>();
if (isColocated()) {
@@ -758,7 +796,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
// 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();
+ removeInvalidRoutes(Env.getCurrentSystemInfo());
ConcurrentHashMap<String, Long> sharedRoutes = new
ConcurrentHashMap<>(primaryClusterToBackend.size());
primaryClusterToBackend.forEach((clusterId, backendId) ->
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 9e92587da5e..ba26adbcc50 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
@@ -28,6 +28,7 @@ import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.TabletMeta;
import org.apache.doris.catalog.TabletSlidingWindowAccessStats;
+import org.apache.doris.cloud.datasource.CloudInternalCatalog;
import org.apache.doris.cloud.persist.UpdateCloudReplicaInfo;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.qe.ComputeGroupException;
@@ -40,6 +41,7 @@ import org.apache.doris.common.ThreadPoolManager;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.rpc.RpcException;
import org.apache.doris.service.FrontendOptions;
@@ -60,6 +62,9 @@ import lombok.Getter;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -86,6 +91,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
private static final int MAX_GLOBAL_TABLET_SET_INITIAL_CAPACITY = 1 << 16;
private static final Function<Long, Set<Long>>
DEFAULT_GLOBAL_TABLET_SET_FACTORY =
ignored -> ConcurrentHashMap.newKeySet();
+ private static final DateTimeFormatter STALE_ROUTE_CLEAN_TIME_FORMAT =
DateTimeFormatter.ofPattern("HH:mm");
private final CloudTabletRebalancerMetrics rebalancerMetrics;
private long currentRoundTabletScanCount;
@@ -1029,24 +1035,62 @@ public class CloudTabletRebalancer extends MasterDaemon
{
// 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) {
+ // Outside the configured window, leave pendingSweepRounds untouched
rather than draining it: a
+ // backend that goes away outside the window must still get its two
rounds once the window
+ // opens, not lose them to rounds that never actually swept.
+ if (pendingSweepRounds > 0 && isStaleRouteCleanTimeAllowed()) {
pendingSweepRounds--;
return true;
}
return false;
}
+ /**
+ * Whether the configured cleanup window
(cloud_tablet_rebalancer_stale_route_clean_start_time to
+ * ..._end_time) contains the current time. Equal start/end -- including
the "00:00"/"00:00" default --
+ * means unrestricted, matching the pre-existing behavior of sweeping
whenever staleRouteSweepNeeded()
+ * says a sweep is due. An unparseable configuration also falls back to
unrestricted rather than
+ * silently disabling cleanup.
+ */
+ @VisibleForTesting
+ boolean isStaleRouteCleanTimeAllowed() {
+ LocalTime start =
parseCleanTime(Config.cloud_tablet_rebalancer_stale_route_clean_start_time);
+ LocalTime end =
parseCleanTime(Config.cloud_tablet_rebalancer_stale_route_clean_end_time);
+ if (start == null || end == null || start.equals(end)) {
+ return true;
+ }
+ return isWithinWindow(LocalTime.now(TimeUtils.getDorisZoneId()),
start, end);
+ }
+
+ @VisibleForTesting
+ static boolean isWithinWindow(LocalTime now, LocalTime start, LocalTime
end) {
+ if (start.isBefore(end)) {
+ return !now.isBefore(start) && !now.isAfter(end);
+ }
+ // across midnight, e.g. 23:00 - 06:00
+ return !now.isBefore(start) || !now.isAfter(end);
+ }
+
+ private static LocalTime parseCleanTime(String time) {
+ try {
+ return LocalTime.parse(time, STALE_ROUTE_CLEAN_TIME_FORMAT);
+ } catch (DateTimeParseException e) {
+ LOG.warn("invalid cloud_tablet_rebalancer_stale_route_clean time:
{}", time);
+ return null;
+ }
+ }
+
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;
+ // Cleanup is independent of compute groups and includes shadow
indices of active tables.
+ // Take the catalog and backend set from the same Env so the sweep can
never mix the two.
+ Env currentEnv = Env.getCurrentEnv();
+ long staleRouteNum = sweepStaleRoutes
+ ? ((CloudInternalCatalog) currentEnv.getInternalCatalog())
+
.removeInvalidCloudReplicaRoutes(currentEnv.getClusterInfo())
+ : 0;
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;
@@ -1058,16 +1102,6 @@ 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
@@ -1137,7 +1171,7 @@ public class CloudTabletRebalancer extends MasterDaemon {
});
LOG.info("collect to editlog route {} infos, error num {}, swept stale
routes {}, entries dropped {}",
- updateReplicaInfos.size(), assignedErrNum[0],
sweepStaleRoutes, staleRouteNum[0]);
+ updateReplicaInfos.size(), assignedErrNum[0],
sweepStaleRoutes, staleRouteNum);
if (updateReplicaInfos.isEmpty()) {
return true;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
index 95228b01c14..574c0df7d88 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java
@@ -62,6 +62,7 @@ import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeConstants;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.util.ColumnsUtil;
import org.apache.doris.common.util.PropertyAnalyzer;
@@ -74,6 +75,7 @@ import org.apache.doris.proto.Types;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.rpc.RpcException;
import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TCompressionType;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.thrift.TSortType;
@@ -1395,6 +1397,83 @@ public class CloudInternalCatalog extends
InternalCatalog {
}
}
+ // The three private helpers below are each independently callable (one
per catalog level), so this
+ // guard is checked in every one of them rather than once at the top -- a
config flip mid-sweep then
+ // takes effect at the next db/table/partition boundary instead of only on
the next whole-catalog call.
+ private static boolean routeCleanupDisabled() {
+ return Config.isNotCloudMode() ||
!Config.enable_cloud_replica_stale_route_clean
+ || FeConstants.runningUnitTest;
+ }
+
+ /**
+ * @param systemInfo the backend set to judge staleness against. Must come
from the same Env this
+ * catalog belongs to (the serving Env during
replay/rebalancing, or the checkpoint's
+ * private Env while generating an image) -- never
resolved internally via
+ * Env.getCurrentSystemInfo(), so a caller cannot
accidentally sweep this catalog's
+ * replicas against a different Env's backend set.
+ */
+ public long removeInvalidCloudReplicaRoutes(SystemInfoService systemInfo) {
+ if (routeCleanupDisabled()) {
+ return 0;
+ }
+ long start = System.currentTimeMillis();
+ long removed = 0;
+ for (Long dbId : getDbIds()) {
+ Database db = getDbNullable(dbId);
+ if (db == null) {
+ continue; // The database can be dropped concurrently on the
serving Env.
+ }
+ removed += removeInvalidCloudReplicaRoutes(db, systemInfo);
+ }
+ LOG.info("swept stale cloud routes, entries dropped {}, cost {} ms",
+ removed, System.currentTimeMillis() - start);
+ return removed;
+ }
+
+ private static long removeInvalidCloudReplicaRoutes(Database db,
SystemInfoService systemInfo) {
+ if (routeCleanupDisabled()) {
+ return 0;
+ }
+ long removed = 0;
+ for (Table table : db.getTables()) {
+ removed += removeInvalidCloudReplicaRoutes(table, systemInfo);
+ }
+ return removed;
+ }
+
+ private static long removeInvalidCloudReplicaRoutes(Table table,
SystemInfoService systemInfo) {
+ if (routeCleanupDisabled() || !table.isManagedTable()) {
+ return 0;
+ }
+ long removed = 0;
+ table.readLock();
+ try {
+ for (Partition partition : ((OlapTable) table).getAllPartitions())
{
+ removed += removeInvalidCloudReplicaRoutes(partition,
systemInfo);
+ }
+ } finally {
+ table.readUnlock();
+ }
+ return removed;
+ }
+
+ private static long removeInvalidCloudReplicaRoutes(Partition partition,
SystemInfoService systemInfo) {
+ if (routeCleanupDisabled()) {
+ return 0;
+ }
+ long removed = 0;
+ for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.ALL, true)) {
+ for (Tablet tablet : index.getTablets()) {
+ for (Replica replica : tablet.getReplicas()) {
+ if (replica instanceof CloudReplica) {
+ removed += ((CloudReplica)
replica).removeInvalidRoutes(systemInfo);
+ }
+ }
+ }
+ }
+ return removed;
+ }
+
private void unprotectUpdateCloudReplica(OlapTable olapTable,
UpdateCloudReplicaInfo info) {
Partition partition = olapTable.getPartition(info.getPartitionId());
if (partition == null) {
@@ -1414,15 +1493,15 @@ public class CloudInternalCatalog extends
InternalCatalog {
Replica replica = tablet.getReplicaById(info.getReplicaId());
Preconditions.checkNotNull(replica, info);
+ CloudSystemInfoService systemInfo = (CloudSystemInfoService)
Env.getCurrentSystemInfo();
String clusterId = info.getClusterId();
- String realClusterId = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
- .getCloudClusterIdByName(clusterId);
+ String realClusterId =
systemInfo.getCloudClusterIdByName(clusterId);
LOG.debug("cluster Id {}, real cluster Id {}", clusterId,
realClusterId);
if (!Strings.isNullOrEmpty(realClusterId)) {
clusterId = realClusterId;
}
- ((CloudReplica) replica).updateClusterToPrimaryBe(clusterId,
info.getBeId());
+ ((CloudReplica)
replica).replayUpdateClusterToPrimaryBe(clusterId, info.getBeId(), systemInfo);
LOG.debug("update single cloud replica cluster {} replica {}
be {}", info.getClusterId(),
replica.getId(), info.getBeId());
@@ -1438,9 +1517,9 @@ public class CloudInternalCatalog extends InternalCatalog
{
}
Preconditions.checkNotNull(replica, info);
+ CloudSystemInfoService systemInfo =
(CloudSystemInfoService) Env.getCurrentSystemInfo();
String clusterId = info.getClusterId();
- String realClusterId = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
- .getCloudClusterIdByName(clusterId);
+ String realClusterId =
systemInfo.getCloudClusterIdByName(clusterId);
LOG.debug("cluster Id {}, real cluster Id {}", clusterId,
realClusterId);
if (!Strings.isNullOrEmpty(realClusterId)) {
clusterId = realClusterId;
@@ -1448,7 +1527,8 @@ public class CloudInternalCatalog extends InternalCatalog
{
LOG.debug("update cloud replica cluster {} replica {} be
{}", info.getClusterId(),
replica.getId(), info.getBeIds().get(i));
- ((CloudReplica)
replica).updateClusterToPrimaryBe(clusterId, info.getBeIds().get(i));
+ ((CloudReplica)
replica).replayUpdateClusterToPrimaryBe(clusterId, info.getBeIds().get(i),
+ systemInfo);
}
}
} catch (Exception e) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index b0f334d354d..3d368a73fc1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -108,6 +108,13 @@ public class CloudSystemInfoService extends
SystemInfoService {
private InstanceInfoPB.Status instanceStatus;
+ // Written and consumed by the journal replay thread; additions never
strand existing routes.
+ private long replayBackendRemovalVersion;
+
+ public long getReplayBackendRemovalVersion() {
+ return replayBackendRemovalVersion;
+ }
+
public long getCloudColocateHrwBeId(GroupId groupId, String clusterId,
List<Long> availableBeIds, long idx) {
return getCloudColocateHrwBeIdInternal(groupId, clusterId,
availableBeIds, idx, -1);
}
@@ -973,6 +980,7 @@ public class CloudSystemInfoService extends
SystemInfoService {
List<Backend> toDel = new ArrayList<>();
toDel.add(backend);
updateCloudClusterMap(new ArrayList<>(), toDel);
+ replayBackendRemovalVersion++;
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
b/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
index 41d86851582..245c643f4c0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
@@ -27,6 +27,7 @@ import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.cloud.catalog.CloudReplica;
+import org.apache.doris.cloud.datasource.CloudInternalCatalog;
import org.apache.doris.common.CheckpointException;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
@@ -44,6 +45,7 @@ import org.apache.doris.persist.Storage;
import org.apache.doris.qe.VariableMgr;
import org.apache.doris.system.Frontend;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -159,6 +161,13 @@ public class Checkpoint extends MasterDaemon {
}
env.postProcessAfterMetadataReplayed(false);
postProcessCloudMetadata();
+ try {
+ removeInvalidCloudReplicaRoutes(env);
+ } catch (Exception e) {
+ // Best effort: a failed sweep only leaves stale routes in
this image, and loading it
+ // runs gsonPostProcess() which cleans them again. Never fail
the checkpoint over it.
+ LOG.warn("failed to sweep stale cloud replica routes before
saving image", e);
+ }
latestImageFilePath = env.saveImage();
replayedJournalId = env.getReplayedJournalId();
@@ -409,6 +418,16 @@ public class Checkpoint extends MasterDaemon {
return lock;
}
+ @VisibleForTesting
+ static long removeInvalidCloudReplicaRoutes(Env env) {
+ if (Config.isNotCloudMode()) {
+ return 0;
+ }
+ // Sweep env's own catalog against env's own backend set -- this env
is the checkpoint's private
+ // Env, so its replicas must never be judged against the serving
cluster's live backends.
+ return ((CloudInternalCatalog)
env.getInternalCatalog()).removeInvalidCloudReplicaRoutes(env.getClusterInfo());
+ }
+
private void postProcessCloudMetadata() {
if (Config.isNotCloudMode()) {
return;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 5929665b25f..de1dd61bb23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -3886,12 +3886,7 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
// This allows the BE to discover peer candidates
across warehouses.
backends = cloudReplica.getAllPrimaryBes();
// Also collect secondary backends (may have cached
data during primary failover)
- for (String computeGroupId :
cloudReplica.getPrimaryComputeGroupIds()) {
- Backend secondaryBe =
cloudReplica.getSecondaryBackend(computeGroupId);
- if (secondaryBe != null) {
- backends.add(secondaryBe);
- }
- }
+ backends.addAll(cloudReplica.getAllSecondaryBes());
} else {
// On the cloud, the PrimaryBackend of a tablet
// indicates the BE where the tablet is stably located,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvRouteCleanupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvRouteCleanupTest.java
new file mode 100644
index 00000000000..4bb3f170452
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvRouteCleanupTest.java
@@ -0,0 +1,145 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 org.apache.doris.cloud.catalog;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.datasource.CloudInternalCatalog;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.Pair;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.ha.FrontendNodeType;
+import org.apache.doris.journal.JournalCursor;
+import org.apache.doris.journal.JournalEntity;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.OperationType;
+import org.apache.doris.resource.Tag;
+import org.apache.doris.system.Backend;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class CloudEnvRouteCleanupTest {
+ private boolean savedClean;
+ private CloudEnv env;
+ private CloudSystemInfoService systemInfo;
+ private CloudInternalCatalog catalog;
+ private MockedStatic<Env> envStatic;
+
+ @BeforeEach
+ public void setUp() {
+ savedClean = Config.enable_cloud_replica_stale_route_clean;
+ Config.enable_cloud_replica_stale_route_clean = true;
+ env = Mockito.mock(CloudEnv.class, Mockito.CALLS_REAL_METHODS);
+ systemInfo = new CloudSystemInfoService();
+ catalog = Mockito.mock(CloudInternalCatalog.class);
+ Mockito.doReturn(systemInfo).when(env).getClusterInfo();
+ Mockito.doReturn(catalog).when(env).getInternalCatalog();
+ Deencapsulation.setField(env, "replayedJournalId", new AtomicLong());
+ Deencapsulation.setField(env, "feType", FrontendNodeType.MASTER);
+ envStatic = Mockito.mockStatic(Env.class);
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfo);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ envStatic.close();
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ }
+
+ @Test
+ public void testCoalescesDropsAtBatchEndEvenWhenFinalTopologyIsUnchanged()
{
+ // Startup and idle batches: exactly one sweep, even without route
update journals.
+ Assertions.assertFalse(env.replayJournal(0));
+ Assertions.assertFalse(env.replayJournal(0));
+ Mockito.verify(catalog,
Mockito.times(1)).removeInvalidCloudReplicaRoutes(systemInfo);
+ Mockito.clearInvocations(catalog);
+
+ Backend first = backend(101L);
+ Backend second = backend(102L);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ JournalCursor cursor = Mockito.mock(JournalCursor.class);
+ Deencapsulation.setField(env, "editLog", editLog);
+ Mockito.when(editLog.read(1L, 4L)).thenReturn(cursor);
+ Mockito.when(cursor.next()).thenReturn(
+ journal(1L, OperationType.OP_ADD_BACKEND, first),
+ journal(2L, OperationType.OP_ADD_BACKEND, second),
+ journal(3L, OperationType.OP_DROP_BACKEND, first),
+ journal(4L, OperationType.OP_DROP_BACKEND, second), null);
+
Mockito.when(catalog.removeInvalidCloudReplicaRoutes(systemInfo)).thenAnswer(invocation
-> {
+ Assertions.assertNull(systemInfo.getBackend(101L));
+ Assertions.assertNull(systemInfo.getBackend(102L));
+ Assertions.assertEquals(4L, env.getReplayedJournalId());
+ return 0L;
+ });
+
+ Assertions.assertTrue(env.replayJournal(4));
+ Assertions.assertEquals(2L,
systemInfo.getReplayBackendRemovalVersion());
+ Assertions.assertFalse(env.replayJournal(4));
+ Mockito.verify(catalog,
Mockito.times(1)).removeInvalidCloudReplicaRoutes(systemInfo);
+
+ // Backend additions alone do not trigger cleanup.
+ systemInfo.replayAddBackend(backend(103L));
+ Assertions.assertFalse(env.replayJournal(4));
+ Mockito.verify(catalog,
Mockito.times(1)).removeInvalidCloudReplicaRoutes(systemInfo);
+ }
+
+ @Test
+ public void testRetriesFailedSweepAndSweepsAfterReenable() {
+ Mockito.when(catalog.removeInvalidCloudReplicaRoutes(systemInfo))
+ .thenThrow(new IllegalStateException("test sweep
failure")).thenReturn(0L);
+ Assertions.assertThrows(IllegalStateException.class, () ->
env.replayJournal(0));
+ Assertions.assertFalse(env.replayJournal(0));
+ Assertions.assertFalse(env.replayJournal(0));
+ Mockito.verify(catalog,
Mockito.times(2)).removeInvalidCloudReplicaRoutes(systemInfo);
+ Config.enable_cloud_replica_stale_route_clean = false;
+ Assertions.assertFalse(env.replayJournal(0));
+ Config.enable_cloud_replica_stale_route_clean = true;
+ Assertions.assertFalse(env.replayJournal(0));
+ Mockito.verify(catalog,
Mockito.times(3)).removeInvalidCloudReplicaRoutes(systemInfo);
+ }
+
+ @Test
+ public void testCheckpointDefersSweepUntilAfterMetadataPostProcessing() {
+ envStatic.when(Env::isCheckpointThread).thenReturn(true);
+ Assertions.assertFalse(env.replayJournal(0));
+ Mockito.verifyNoInteractions(catalog);
+ }
+
+ private static Backend backend(long id) {
+ Backend backend = new Backend(id, "127.0.0.1", (int) (9050 + id));
+ backend.setTagMap(Map.of(Tag.TYPE_LOCATION, "default",
+ Tag.CLOUD_CLUSTER_ID, "cg", Tag.CLOUD_CLUSTER_NAME, "cg"));
+ return backend;
+ }
+
+ private static Pair<Long, JournalEntity> journal(long id, short opCode,
Backend backend) {
+ JournalEntity entity = new JournalEntity();
+ entity.setOpCode(opCode);
+ entity.setData(backend);
+ return Pair.of(id, entity);
+ }
+}
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 cc2c05c3951..ddbfddcc5b1 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
@@ -467,7 +467,7 @@ public class CloudReplicaTest {
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(2,
replica.removeInvalidRoutes(mockInfoService));
Assertions.assertEquals(1,
replica.getPrimaryComputeGroupIds().size());
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1));
@@ -494,7 +494,7 @@ public class CloudReplicaTest {
replica.updateClusterToPrimaryBe(CLUSTER_ID_2, 2001L);
replica.updateClusterToSecondaryBe(CLUSTER_ID_2, 2001L);
- Assertions.assertEquals(0, replica.removeInvalidRoutes());
+ Assertions.assertEquals(0,
replica.removeInvalidRoutes(mockInfoService));
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_2));
Backend revivedBe = createBackend(2001L, true, false);
@@ -523,7 +523,7 @@ public class CloudReplicaTest {
replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 2001L);
replica.updateClusterToSecondaryBe(CLUSTER_ID_1, 1001L);
- Assertions.assertEquals(0, replica.removeInvalidRoutes());
+ Assertions.assertEquals(0,
replica.removeInvalidRoutes(mockInfoService));
// FrontendServiceImpl's lazy fetch reaches the secondary through
the primary key set
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(CLUSTER_ID_1));
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 f2b630947f1..d58ae18f624 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
@@ -27,10 +27,11 @@ import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.TabletInvertedIndex;
import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.cloud.datasource.CloudInternalCatalog;
import org.apache.doris.cloud.persist.UpdateCloudReplicaInfo;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
-import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
@@ -46,6 +47,8 @@ import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
import java.util.AbstractList;
import java.util.AbstractMap;
import java.util.ArrayList;
@@ -365,6 +368,7 @@ public class CloudTabletRebalancerTest {
Long beId = 60_001L;
String clusterId = "cluster-a";
setField(rebalancer, "clusterToBes",
Collections.singletonMap(clusterId, List.of(beId)));
+ setField(rebalancer, "allBes", Set.of(beId));
try (MockedStatic<Env> ignored = mockRouteEnvironment(
dbId, tableId, partitionId, indexId, tabletId, clusterId,
beId, 3)) {
@@ -387,6 +391,7 @@ public class CloudTabletRebalancerTest {
Long beId = 60_001L;
String clusterId = "cluster-a";
setField(rebalancer, "clusterToBes",
Collections.singletonMap(clusterId, List.of(beId)));
+ setField(rebalancer, "allBes", Set.of(beId));
try (MockedStatic<Env> ignored = mockRouteEnvironment(
dbId, tableId, partitionId, indexId, tabletId, clusterId,
beId, 3, true)) {
@@ -829,8 +834,9 @@ public class CloudTabletRebalancerTest {
Env env = Mockito.mock(Env.class);
TabletInvertedIndex invertedIndex =
Mockito.mock(TabletInvertedIndex.class);
TabletMeta tabletMeta = Mockito.mock(TabletMeta.class);
- InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ CloudInternalCatalog catalog = Mockito.spy(new CloudInternalCatalog());
Mockito.when(env.getTabletInvertedIndex()).thenReturn(invertedIndex);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
Mockito.when(invertedIndex.getTabletMeta(tabletId)).thenReturn(tabletMeta);
Mockito.when(tabletMeta.getTableId()).thenReturn(tableId);
Mockito.when(tabletMeta.getPartitionId()).thenReturn(partitionId);
@@ -876,7 +882,7 @@ public class CloudTabletRebalancerTest {
Env env = Mockito.mock(Env.class);
TabletInvertedIndex invertedIndex =
Mockito.mock(TabletInvertedIndex.class);
TabletMeta tabletMeta = Mockito.mock(TabletMeta.class);
- InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ CloudInternalCatalog catalog = Mockito.spy(new CloudInternalCatalog());
ColocateTableIndex colocateTableIndex =
Mockito.mock(ColocateTableIndex.class);
Database database = Mockito.mock(Database.class);
OlapTable table = Mockito.mock(OlapTable.class);
@@ -888,6 +894,8 @@ public class CloudTabletRebalancerTest {
Long tabletId = tablet.getId();
Mockito.when(env.getTabletInvertedIndex()).thenReturn(invertedIndex);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ Mockito.when(env.getClusterInfo()).thenReturn(systemInfoService);
Mockito.when(invertedIndex.getTabletMeta(tabletId)).thenReturn(tabletMeta);
Mockito.when(tabletMeta.getTableId()).thenReturn(tableId);
Mockito.when(tabletMeta.getPartitionId()).thenReturn(partitionId);
@@ -1251,6 +1259,15 @@ public class CloudTabletRebalancerTest {
// a backend that disappeared triggers a sweep
Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L))));
+ // Cleanup must also run after the last compute group disappears.
+ TestRebalancer lastGroupRebalancer = new TestRebalancer();
+
Assertions.assertTrue(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L))));
+
Assertions.assertTrue(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(1L))));
+
Assertions.assertTrue(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>()));
+
Assertions.assertTrue(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>()));
+
Assertions.assertFalse(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>()));
+
Assertions.assertFalse(lastGroupRebalancer.staleRouteSweepNeeded(new
HashSet<>(Arrays.asList(2L))));
+
// 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));
@@ -1260,4 +1277,100 @@ public class CloudTabletRebalancerTest {
Config.enable_cloud_replica_stale_route_clean = saved;
}
}
+
+ @Test
+ public void testIsWithinWindow() {
+ // normal (non-wrapping) window
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(12, 0), LocalTime.of(10, 0), LocalTime.of(14,
0)));
+ Assertions.assertFalse(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(9, 59), LocalTime.of(10, 0), LocalTime.of(14,
0)));
+ Assertions.assertFalse(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(14, 1), LocalTime.of(10, 0), LocalTime.of(14,
0)));
+ // boundaries are inclusive
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(10, 0), LocalTime.of(10, 0), LocalTime.of(14,
0)));
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(14, 0), LocalTime.of(10, 0), LocalTime.of(14,
0)));
+
+ // window across midnight, e.g. 23:00 - 06:00
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(23, 30), LocalTime.of(23, 0), LocalTime.of(6,
0)));
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(1, 0), LocalTime.of(23, 0), LocalTime.of(6, 0)));
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(23, 0), LocalTime.of(23, 0), LocalTime.of(6, 0)));
+ Assertions.assertTrue(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(6, 0), LocalTime.of(23, 0), LocalTime.of(6, 0)));
+ Assertions.assertFalse(CloudTabletRebalancer.isWithinWindow(
+ LocalTime.of(12, 0), LocalTime.of(23, 0), LocalTime.of(6, 0)));
+ }
+
+ @Test
+ public void testIsStaleRouteCleanTimeAllowedDefaultsToUnrestricted() {
+ String savedStart =
Config.cloud_tablet_rebalancer_stale_route_clean_start_time;
+ String savedEnd =
Config.cloud_tablet_rebalancer_stale_route_clean_end_time;
+ try {
+ TestRebalancer rebalancer = new TestRebalancer();
+
+ // the actual default
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
"00:00";
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
"00:00";
+ Assertions.assertTrue(rebalancer.isStaleRouteCleanTimeAllowed());
+
+ // any equal start/end means unrestricted, not just the default
value
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
"09:30";
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
"09:30";
+ Assertions.assertTrue(rebalancer.isStaleRouteCleanTimeAllowed());
+
+ // an unparseable configuration falls back to unrestricted rather
than disabling cleanup
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
"not-a-time";
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
"18:00";
+ Assertions.assertTrue(rebalancer.isStaleRouteCleanTimeAllowed());
+ } finally {
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
savedStart;
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
savedEnd;
+ }
+ }
+
+ @Test
+ public void testStaleRouteSweepGateRespectsCleanWindow() {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ String savedStart =
Config.cloud_tablet_rebalancer_stale_route_clean_start_time;
+ String savedEnd =
Config.cloud_tablet_rebalancer_stale_route_clean_end_time;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
"00:00";
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
"00:00";
+ TestRebalancer rebalancer = new TestRebalancer();
+ Set<Long> bes = new HashSet<>(Arrays.asList(1L, 2L));
+ // establish a clean baseline with no rounds pending
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes));
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(bes));
+
+ // close the window (a 1-2 hour offset from "now" is never in it),
then a backend disappears
+ DateTimeFormatter fmt = DateTimeFormatter.ofPattern("HH:mm");
+ LocalTime now = LocalTime.now(TimeUtils.getDorisZoneId());
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
now.plusHours(1).format(fmt);
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
now.plusHours(2).format(fmt);
+ Set<Long> shrunk = new HashSet<>(Arrays.asList(1L));
+ // the drop is detected (baseline advances) but the window blocks
the sweep -- and the two
+ // rounds it owes must not be drained by rounds that never
actually ran
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(shrunk));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(shrunk));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(shrunk));
+
+ // window opens: the two rounds owed from the earlier drop are
still there, untouched
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
"00:00";
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
"00:00";
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(shrunk));
+ Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(shrunk));
+ Assertions.assertFalse(rebalancer.staleRouteSweepNeeded(shrunk));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ Config.cloud_tablet_rebalancer_stale_route_clean_start_time =
savedStart;
+ Config.cloud_tablet_rebalancer_stale_route_clean_end_time =
savedEnd;
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java
index 6f124c7fad9..cabc592f4d3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java
@@ -18,22 +18,35 @@
package org.apache.doris.cloud.datasource;
import org.apache.doris.analysis.DataSortInfo;
+import org.apache.doris.catalog.ColocateTableIndex;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Tablet;
+import org.apache.doris.cloud.catalog.CloudReplica;
+import org.apache.doris.cloud.catalog.CloudTablet;
+import org.apache.doris.cloud.persist.UpdateCloudReplicaInfo;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.rpc.MetaServiceProxy;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeConstants;
import org.apache.doris.proto.OlapFile;
+import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TCompressionType;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.thrift.TSortType;
import org.apache.doris.thrift.TStorageFormat;
import org.apache.doris.thrift.TTabletType;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
@@ -44,6 +57,45 @@ import java.util.Collections;
import java.util.List;
public class CloudInternalCatalogTest {
+ private static final long DB_ID = 10001L;
+ private static final long TABLE_ID = 20001L;
+ private static final long PARTITION_ID = 30001L;
+ private static final long INDEX_ID = 40001L;
+ private static final long LIVE_BE_ID = 50001L;
+ private static final long DEAD_BE_ID = 50002L;
+ private static final String STALE_CLUSTER_ID = "stale_cluster";
+ private static final String LIVE_CLUSTER_ID = "live_cluster";
+
+ private MockedStatic<Env> envMockedStatic;
+ private CloudSystemInfoService systemInfoService;
+ private CloudInternalCatalog catalog;
+ private MaterializedIndex materializedIndex;
+
+ @BeforeEach
+ public void setUpReplayCatalog() {
+ systemInfoService = Mockito.mock(CloudSystemInfoService.class);
+ ColocateTableIndex colocateTableIndex =
Mockito.mock(ColocateTableIndex.class);
+ envMockedStatic = Mockito.mockStatic(Env.class);
+
envMockedStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
envMockedStatic.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex);
+
Mockito.when(colocateTableIndex.isColocateTableNoLock(TABLE_ID)).thenReturn(false);
+
+ catalog = Mockito.spy(new CloudInternalCatalog());
+ Database database = Mockito.mock(Database.class);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Partition partition = Mockito.mock(Partition.class);
+ materializedIndex = Mockito.mock(MaterializedIndex.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable(DB_ID);
+ Mockito.when(database.getTableNullable(TABLE_ID)).thenReturn(table);
+ Mockito.when(table.getPartition(PARTITION_ID)).thenReturn(partition);
+
Mockito.when(partition.getIndex(INDEX_ID)).thenReturn(materializedIndex);
+ }
+
+ @AfterEach
+ public void tearDownReplayCatalog() {
+ envMockedStatic.close();
+ }
+
@Test
public void testCreateTabletMetaUsesCurrentSchemaVersionAndFormat() throws
Exception {
Tablet tablet = Mockito.mock(Tablet.class);
@@ -114,6 +166,138 @@ public class CloudInternalCatalogTest {
}
}
+ @Test
+ public void testReplaySingleReplicaDoesNotScanOtherRoutes() throws
Exception {
+ 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;
+ setBackend(LIVE_BE_ID, Mockito.mock(Backend.class));
+ setBackend(DEAD_BE_ID, null);
+ CloudReplica replica = addReplayReplica(60001L, 70001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ catalog.replayUpdateCloudReplica(new UpdateCloudReplicaInfo(DB_ID,
TABLE_ID, PARTITION_ID, INDEX_ID,
+ 60001L, 70001L, LIVE_CLUSTER_ID, LIVE_BE_ID));
+
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Mockito.verify(replica,
Mockito.never()).removeInvalidRoutes(Mockito.any());
+ Assertions.assertEquals(LIVE_BE_ID,
replica.getClusterPrimaryBackendId(LIVE_CLUSTER_ID));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void
testReplayBatchDoesNotScanOtherRoutesWithAndWithoutReplicaIds() throws
Exception {
+ 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;
+ setBackend(LIVE_BE_ID, Mockito.mock(Backend.class));
+ setBackend(DEAD_BE_ID, null);
+ List<CloudReplica> replicas = Arrays.asList(
+ addReplayReplica(60001L, 70001L), addReplayReplica(60002L,
70002L));
+ replicas.forEach(replica ->
replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID));
+ List<Long> tabletIds = Arrays.asList(60001L, 60002L);
+ List<Long> beIds = Arrays.asList(LIVE_BE_ID, LIVE_BE_ID);
+
+ UpdateCloudReplicaInfo explicitReplicaIds = new
UpdateCloudReplicaInfo(
+ DB_ID, TABLE_ID, PARTITION_ID, INDEX_ID, LIVE_CLUSTER_ID,
beIds, tabletIds);
+ explicitReplicaIds.setReplicaIds(Arrays.asList(70001L, 70002L));
+ catalog.replayUpdateCloudReplica(explicitReplicaIds);
+
+ for (CloudReplica replica : replicas) {
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Assertions.assertEquals(LIVE_BE_ID,
replica.getClusterPrimaryBackendId(LIVE_CLUSTER_ID));
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+ }
+
+ catalog.replayUpdateCloudReplica(new UpdateCloudReplicaInfo(
+ DB_ID, TABLE_ID, PARTITION_ID, INDEX_ID, LIVE_CLUSTER_ID,
beIds, tabletIds));
+
+ for (CloudReplica replica : replicas) {
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Mockito.verify(replica,
Mockito.never()).removeInvalidRoutes(Mockito.any());
+ Assertions.assertEquals(LIVE_BE_ID,
replica.getClusterPrimaryBackendId(LIVE_CLUSTER_ID));
+ }
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testReplayKeepsStaleRoutesWhenCleanupDisabled() throws
Exception {
+ 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;
+ setBackend(LIVE_BE_ID, Mockito.mock(Backend.class));
+ setBackend(DEAD_BE_ID, null);
+ CloudReplica replica = addReplayReplica(60001L, 70001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ catalog.replayUpdateCloudReplica(new UpdateCloudReplicaInfo(DB_ID,
TABLE_ID, PARTITION_ID, INDEX_ID,
+ 60001L, 70001L, LIVE_CLUSTER_ID, LIVE_BE_ID));
+
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Assertions.assertEquals(LIVE_BE_ID,
replica.getClusterPrimaryBackendId(LIVE_CLUSTER_ID));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ @Test
+ public void testReplayRemovesDeletedBackendRouteAndKeepsLiveSecondary()
throws Exception {
+ 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 liveBackend = Mockito.mock(Backend.class);
+ setBackend(LIVE_BE_ID, liveBackend);
+ setBackend(DEAD_BE_ID, null);
+ CloudReplica replica = addReplayReplica(60001L, 70001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+ replica.updateClusterToSecondaryBe(STALE_CLUSTER_ID, LIVE_BE_ID);
+
+ catalog.replayUpdateCloudReplica(new UpdateCloudReplicaInfo(DB_ID,
TABLE_ID, PARTITION_ID, INDEX_ID,
+ 60001L, 70001L, LIVE_CLUSTER_ID, DEAD_BE_ID));
+
+
Assertions.assertFalse(replica.getPrimaryComputeGroupIds().contains(LIVE_CLUSTER_ID));
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Assertions.assertSame(liveBackend,
replica.getSecondaryBackend(STALE_CLUSTER_ID));
+
+ // A late journal must not clear the valid fallback for this same
compute group.
+ catalog.replayUpdateCloudReplica(new UpdateCloudReplicaInfo(DB_ID,
TABLE_ID, PARTITION_ID, INDEX_ID,
+ 60001L, 70001L, STALE_CLUSTER_ID, DEAD_BE_ID));
+ Assertions.assertSame(liveBackend,
replica.getSecondaryBackend(STALE_CLUSTER_ID));
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ }
+ }
+
+ private CloudReplica addReplayReplica(long tabletId, long replicaId) {
+ CloudReplica replica = Mockito.spy(new CloudReplica(replicaId, -1L,
Replica.ReplicaState.NORMAL, 1L, 0,
+ DB_ID, TABLE_ID, PARTITION_ID, INDEX_ID, 0L));
+ CloudTablet tablet = new CloudTablet(tabletId);
+ tablet.addReplica(replica, true);
+ Mockito.when(materializedIndex.getTablet(tabletId)).thenReturn(tablet);
+ return replica;
+ }
+
+ private void setBackend(long backendId, Backend backend) {
+
Mockito.when(systemInfoService.getBackendByIdWithBoxedId(backendId)).thenReturn(backend);
+
Mockito.when(systemInfoService.getBackend(backendId)).thenReturn(backend);
+ }
+
private OlapFile.TabletMetaCloudPB createTabletMeta(Tablet tablet) throws
Exception {
return new CloudInternalCatalog().createTabletMetaBuilder(
1L, 2L, 3L, tablet, TTabletType.TABLET_TYPE_DISK, 23,
KeysType.DUP_KEYS, (short) 1,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/master/CheckpointTest.java
b/fe/fe-core/src/test/java/org/apache/doris/master/CheckpointTest.java
new file mode 100644
index 00000000000..1ddd31e07cd
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/master/CheckpointTest.java
@@ -0,0 +1,276 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 org.apache.doris.master;
+
+import org.apache.doris.catalog.CatalogRecycleBin;
+import org.apache.doris.catalog.ColocateTableIndex;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.catalog.CloudReplica;
+import org.apache.doris.cloud.catalog.CloudTablet;
+import org.apache.doris.cloud.datasource.CloudInternalCatalog;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.system.Backend;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+public class CheckpointTest {
+ private static final long DB_ID = 10001L;
+ private static final long TABLE_ID = 20001L;
+ private static final long PARTITION_ID = 30001L;
+ private static final long INDEX_ID = 40001L;
+ private static final long LIVE_BE_ID = 50001L;
+ private static final long DEAD_BE_ID = 50002L;
+ private static final String STALE_CLUSTER_ID = "stale_cluster";
+ private static final String LIVE_CLUSTER_ID = "live_cluster";
+
+ private MockedStatic<Env> envMockedStatic;
+ private Env env;
+ private Database database;
+ private OlapTable table;
+ private Partition partition;
+ private MaterializedIndex materializedIndex;
+ private Backend liveBackend;
+ private CatalogRecycleBin recycleBin;
+
+ @BeforeEach
+ public void setUp() {
+ CloudSystemInfoService systemInfoService =
Mockito.mock(CloudSystemInfoService.class);
+ ColocateTableIndex colocateTableIndex =
Mockito.mock(ColocateTableIndex.class);
+ envMockedStatic = Mockito.mockStatic(Env.class);
+
envMockedStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
envMockedStatic.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex);
+
Mockito.when(colocateTableIndex.isColocateTableNoLock(TABLE_ID)).thenReturn(false);
+ liveBackend = Mockito.mock(Backend.class);
+
Mockito.when(systemInfoService.getBackendByIdWithBoxedId(LIVE_BE_ID)).thenReturn(liveBackend);
+
Mockito.when(systemInfoService.getBackendByIdWithBoxedId(DEAD_BE_ID)).thenReturn(null);
+
Mockito.when(systemInfoService.getBackend(LIVE_BE_ID)).thenReturn(liveBackend);
+
+ env = Mockito.mock(Env.class);
+ CloudInternalCatalog catalog = Mockito.spy(new CloudInternalCatalog());
+ database = Mockito.mock(Database.class);
+ table = Mockito.mock(OlapTable.class);
+ partition = Mockito.mock(Partition.class);
+ materializedIndex = Mockito.mock(MaterializedIndex.class);
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+ recycleBin = new CatalogRecycleBin();
+ envMockedStatic.when(Env::getCurrentRecycleBin).thenReturn(recycleBin);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ Mockito.when(env.getClusterInfo()).thenReturn(systemInfoService);
+
Mockito.when(catalog.getDbIds()).thenReturn(Collections.singletonList(DB_ID));
+ Mockito.when(catalog.getDbNullable(DB_ID)).thenReturn(database);
+
Mockito.when(database.getTables()).thenReturn(Collections.singletonList(table));
+ Mockito.when(table.isManagedTable()).thenReturn(true);
+
Mockito.when(table.getAllPartitions()).thenReturn(Collections.singletonList(partition));
+ Mockito.when(partition.getMaterializedIndices(IndexExtState.ALL, true))
+ .thenReturn(Collections.singletonList(materializedIndex));
+ }
+
+ @AfterEach
+ public void tearDown() {
+ envMockedStatic.close();
+ }
+
+ @Test
+ public void testDoesNotSweepRecycledTableAndPartition() {
+ withRouteCleanup(() -> {
+ CloudReplica tableReplica = addReplica(materializedIndex, 60001L);
+ tableReplica.updateClusterToPrimaryBe(STALE_CLUSTER_ID,
DEAD_BE_ID);
+
Mockito.when(database.getTables()).thenReturn(Collections.emptyList());
+ Mockito.when(table.getId()).thenReturn(TABLE_ID);
+ Mockito.when(table.getName()).thenReturn("recycled_table");
+ Assertions.assertTrue(recycleBin.recycleTable(DB_ID, table, true,
false, 1L));
+ Partition recycledPartition = Mockito.mock(Partition.class);
+ Mockito.when(recycledPartition.getId()).thenReturn(PARTITION_ID +
1);
+
Mockito.when(recycledPartition.getName()).thenReturn("recycled_partition");
+ MaterializedIndex recycledIndex =
Mockito.mock(MaterializedIndex.class);
+
Mockito.when(recycledPartition.getMaterializedIndices(IndexExtState.ALL, true))
+ .thenReturn(Collections.singletonList(recycledIndex));
+ CloudReplica partitionReplica = addReplica(recycledIndex, 60002L);
+ partitionReplica.updateClusterToPrimaryBe(STALE_CLUSTER_ID,
DEAD_BE_ID);
+ Assertions.assertTrue(recycleBin.recyclePartition(DB_ID, TABLE_ID,
"table", recycledPartition,
+ null, null, null, null, false, true));
+
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertTrue(tableReplica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+
Assertions.assertTrue(partitionReplica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ envMockedStatic.verify(Env::getCurrentRecycleBin, Mockito.never());
+ });
+ }
+
+ @Test
+ public void testRemovesStaleRouteAndKeepsLiveRoute() {
+ withRouteCleanup(() -> {
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+ replica.updateClusterToPrimaryBe(LIVE_CLUSTER_ID, LIVE_BE_ID);
+
+ Assertions.assertEquals(1L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertFalse(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Assertions.assertEquals(LIVE_BE_ID,
replica.getClusterPrimaryBackendId(LIVE_CLUSTER_ID));
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+ });
+ }
+
+ @Test
+ public void testKeepsDeadPrimaryWithLiveSecondary() {
+ withRouteCleanup(() -> {
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+ replica.updateClusterToSecondaryBe(STALE_CLUSTER_ID, LIVE_BE_ID);
+
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+ Assertions.assertEquals(DEAD_BE_ID,
replica.getClusterPrimaryBackendId(STALE_CLUSTER_ID));
+ Assertions.assertSame(liveBackend,
replica.getSecondaryBackend(STALE_CLUSTER_ID));
+ });
+ }
+
+ @Test
+ public void testNonCloudModeIsNoOp() {
+ withRouteCleanup(() -> {
+ Config.deploy_mode = "local";
+ Config.cloud_unique_id = "";
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ Mockito.verify(env, Mockito.never()).getInternalCatalog();
+ });
+ }
+
+ @Test
+ public void testSkipsNonManagedTable() {
+ withRouteCleanup(() -> {
+ Table nonManagedTable = Mockito.mock(Table.class);
+ Mockito.when(nonManagedTable.isManagedTable()).thenReturn(false);
+
Mockito.when(database.getTables()).thenReturn(Arrays.asList(nonManagedTable,
table));
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ Assertions.assertEquals(1L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertFalse(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ });
+ }
+
+ @Test
+ public void testSweepsShadowIndex() {
+ withRouteCleanup(() -> {
+ MaterializedIndex shadowIndex =
Mockito.mock(MaterializedIndex.class);
+
Mockito.when(shadowIndex.getState()).thenReturn(MaterializedIndex.IndexState.SHADOW);
+
Mockito.when(partition.getMaterializedIndices(IndexExtState.VISIBLE, true))
+ .thenReturn(Collections.singletonList(materializedIndex));
+ Mockito.when(partition.getMaterializedIndices(IndexExtState.ALL,
true))
+ .thenReturn(Arrays.asList(materializedIndex, shadowIndex));
+ CloudReplica visibleReplica = addReplica(materializedIndex,
60001L);
+ CloudReplica shadowReplica = addReplica(shadowIndex, 60002L);
+ visibleReplica.updateClusterToPrimaryBe(STALE_CLUSTER_ID,
DEAD_BE_ID);
+ shadowReplica.updateClusterToPrimaryBe(STALE_CLUSTER_ID,
DEAD_BE_ID);
+
+ Assertions.assertEquals(2L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertFalse(visibleReplica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+
Assertions.assertFalse(shadowReplica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ });
+ }
+
+ @Test
+ public void testSweepsAgainstCheckpointEnvBackendsNotServingEnvBackends() {
+ withRouteCleanup(() -> {
+ // The serving Env (reached via the static
Env.getCurrentSystemInfo()) still thinks DEAD_BE_ID
+ // is alive -- only the checkpoint's own private Env
(env.getClusterInfo()) has it removed. If
+ // the sweep ever regresses to resolving backends ambiently
instead of from `env`, this route
+ // would wrongly survive.
+ CloudSystemInfoService servingSystemInfo =
Mockito.mock(CloudSystemInfoService.class);
+ Backend stillAliveOnServingEnv = Mockito.mock(Backend.class);
+
Mockito.when(servingSystemInfo.getBackendByIdWithBoxedId(DEAD_BE_ID)).thenReturn(stillAliveOnServingEnv);
+
envMockedStatic.when(Env::getCurrentSystemInfo).thenReturn(servingSystemInfo);
+
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ Assertions.assertEquals(1L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertFalse(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ });
+ }
+
+ @Test
+ public void testSkipsNonCloudReplica() {
+ withRouteCleanup(() -> {
+ CloudTablet tablet = new CloudTablet(60001L);
+ tablet.addReplica(Mockito.mock(Replica.class), true);
+
Mockito.when(materializedIndex.getTablets()).thenReturn(Collections.singletonList(tablet));
+
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+ });
+ }
+
+ @Test
+ public void testCleanupDisabledIsNoOp() {
+ withRouteCleanup(() -> {
+ Config.enable_cloud_replica_stale_route_clean = false;
+ CloudReplica replica = addReplica(materializedIndex, 60001L);
+ replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID);
+
+ Assertions.assertEquals(0L,
Checkpoint.removeInvalidCloudReplicaRoutes(env));
+
Assertions.assertTrue(replica.getPrimaryComputeGroupIds().contains(STALE_CLUSTER_ID));
+ });
+ }
+
+ private CloudReplica addReplica(MaterializedIndex index, long tabletId) {
+ CloudReplica replica = new CloudReplica(tabletId + 10000L, -1L,
Replica.ReplicaState.NORMAL, 1L, 0,
+ DB_ID, TABLE_ID, PARTITION_ID, INDEX_ID, 0L);
+ CloudTablet tablet = new CloudTablet(tabletId);
+ tablet.addReplica(replica, true);
+
Mockito.when(index.getTablets()).thenReturn(Collections.singletonList(tablet));
+ return replica;
+ }
+
+ private void withRouteCleanup(Runnable test) {
+ boolean savedClean = Config.enable_cloud_replica_stale_route_clean;
+ boolean savedUnitTest = FeConstants.runningUnitTest;
+ String savedDeployMode = Config.deploy_mode;
+ String savedCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.enable_cloud_replica_stale_route_clean = true;
+ FeConstants.runningUnitTest = false;
+ Config.deploy_mode = "cloud";
+ test.run();
+ } finally {
+ Config.enable_cloud_replica_stale_route_clean = savedClean;
+ FeConstants.runningUnitTest = savedUnitTest;
+ Config.deploy_mode = savedDeployMode;
+ Config.cloud_unique_id = savedCloudUniqueId;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java
index ef0cd3e7be4..eefcfc29795 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java
@@ -17,10 +17,16 @@
package org.apache.doris.service;
+import org.apache.doris.catalog.ColocateTableIndex;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.TabletInvertedIndex;
import org.apache.doris.cloud.CacheHotspotManager;
import org.apache.doris.cloud.catalog.CloudEnv;
+import org.apache.doris.cloud.catalog.CloudReplica;
import org.apache.doris.common.Config;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TGetTabletReplicaInfosRequest;
import org.apache.doris.thrift.TGetTabletReplicaInfosResult;
import org.apache.doris.thrift.TStatusCode;
@@ -34,6 +40,35 @@ import java.util.Collections;
public class FrontendServiceImplCloudTest {
+ @Test
+ public void testSecondaryOnlyRouteIsDiscovered() {
+ String saved = Config.cloud_unique_id;
+ Config.cloud_unique_id = "secondary-peer-test";
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class)) {
+
envMock.when(Env::getCurrentEnv).thenReturn(Mockito.mock(CloudEnv.class));
+
envMock.when(Env::getCurrentColocateIndex).thenReturn(Mockito.mock(ColocateTableIndex.class));
+ SystemInfoService systemInfo =
Mockito.mock(SystemInfoService.class);
+ envMock.when(Env::getCurrentSystemInfo).thenReturn(systemInfo);
+ Backend backend = new Backend(12L, "127.0.0.1", 9050);
+ Mockito.when(systemInfo.getBackend(12L)).thenReturn(backend);
+
Mockito.when(systemInfo.getBackendByIdWithBoxedId(12L)).thenReturn(backend);
+ CloudReplica replica = new CloudReplica(1L, -1L,
Replica.ReplicaState.NORMAL, 1L, 0,
+ 2L, 3L, 4L, 5L, 0L);
+ // Cleanup and secondary publication can leave this valid
secondary without a primary key.
+ replica.updateClusterToSecondaryBe("cg", 12L);
+ TabletInvertedIndex index =
Mockito.mock(TabletInvertedIndex.class);
+ envMock.when(Env::getCurrentInvertedIndex).thenReturn(index);
+
Mockito.when(index.getReplicasByTabletId(789L)).thenReturn(Collections.singletonList(replica));
+ TGetTabletReplicaInfosRequest request = new
TGetTabletReplicaInfosRequest();
+ request.setTabletIds(Collections.singletonList(789L));
+ TGetTabletReplicaInfosResult result = new
FrontendServiceImpl(Mockito.mock(ExecuteEnv.class))
+ .getTabletReplicaInfos(request);
+ Assertions.assertEquals(1,
result.getTabletReplicaInfos().get(789L).size());
+ } finally {
+ Config.cloud_unique_id = saved;
+ }
+ }
+
// Regression test for FrontendServiceImpl.getTabletReplicaInfos NPE:
// When a warm-up job has been removed from
// CacheHotspotManager.cloudWarmUpJobs (past
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
