From 1f7c35c74672f8708bd89e6f38bd3b74640e4c26 Mon Sep 17 00:00:00 2001 From: deardeng Date: Wed, 14 Oct 2026 15:16:44 +0800 Subject: [PATCH] [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. - 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. 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. --- .../java/org/apache/doris/common/Config.java | 21 +- .../apache/doris/cloud/catalog/CloudEnv.java | 27 ++ .../doris/cloud/catalog/CloudReplica.java | 58 +++- .../cloud/catalog/CloudTabletRebalancer.java | 72 +++-- .../datasource/CloudInternalCatalog.java | 92 +++++- .../cloud/system/CloudSystemInfoService.java | 8 + .../org/apache/doris/master/Checkpoint.java | 19 ++ .../doris/service/FrontendServiceImpl.java | 7 +- .../catalog/CloudEnvRouteCleanupTest.java | 145 +++++++++ .../doris/cloud/catalog/CloudReplicaTest.java | 89 ++++++ .../catalog/CloudTabletRebalancerTest.java | 144 +++++++++ .../datasource/CloudInternalCatalogTest.java | 184 ++++++++++++ .../apache/doris/master/CheckpointTest.java | 276 ++++++++++++++++++ .../service/FrontendServiceImplCloudTest.java | 70 +++++ 14 files changed, 1168 insertions(+), 44 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvRouteCleanupTest.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/master/CheckpointTest.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java 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 c311f4c55ec483..303aae3c4e6cb9 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 @@ -3889,11 +3889,26 @@ public void handle(Field field, String value) throws Exception { @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. " 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 85103b73487168..681283576a6eed 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 4c4bcb175a8f21..3c9465ced8a0ec 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 @@ -417,8 +417,11 @@ public boolean discardRouteIfBackendVanished(long beId) { 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; @@ -647,13 +650,17 @@ public void clearClusterToBe(String cluster) { * 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. @@ -663,18 +670,24 @@ public int removeInvalidRoutes() { if (systemInfo.getBackend(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 entry : primaryClusterToBackend.entrySet()) { if (systemInfo.getBackend(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; @@ -682,7 +695,6 @@ public int removeInvalidRoutes() { /** * 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 getPrimaryComputeGroupIds() { return primaryClusterToBackend.keySet(); @@ -707,6 +719,32 @@ public void checkAndClearSecondaryClusterToBe(String clusterId, long expireTimes } } + public List getAllSecondaryBes() { + List result = new ArrayList<>(); + SystemInfoService systemInfo = Env.getCurrentSystemInfo(); + for (Pair 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 getAllPrimaryBes() { List result = new ArrayList(); if (isColocated()) { @@ -757,7 +795,7 @@ public void gsonPostProcess() throws IOException { // 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 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 3519b579b53813..180e8eb6e2bb51 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.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.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 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> 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; @@ -1005,24 +1011,62 @@ boolean staleRouteSweepNeeded(Set currentBes) { // 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 updateReplicaInfos = new ArrayList(); long[] assignedErrNum = {0L}; - long[] staleRouteNum = {0L}; boolean sweepStaleRoutes = staleRouteSweepNeeded(allBes); - // loopCloudReplica() has the compute group loop innermost, so it hands us every replica once per - // live compute group, while removeInvalidRoutes() scans the whole route map and does not care - // which group we are on. Pin the sweep to one arbitrary group id so a sweeping round still makes a - // single pass per replica. If clusterToBes is empty the callback never runs at all, so the serving - // catalog keeps the entries until it reloads the image -- there is nothing to route in that state. - String sweepTicket = sweepStaleRoutes ? clusterToBes.keySet().stream().findFirst().orElse(null) : null; + // 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; @@ -1034,16 +1078,6 @@ private boolean completeRouteInfo() { 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 @@ -1113,7 +1147,7 @@ private boolean completeRouteInfo() { }); 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 7bf95c19d1504f..8d1a5a68385e52 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 @@ -57,6 +57,7 @@ import org.apache.doris.cluster.ClusterNamespace; 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.datasource.InternalCatalog; import org.apache.doris.proto.OlapCommon; @@ -66,6 +67,7 @@ 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; @@ -1162,6 +1164,83 @@ public void replayUpdateCloudReplica(UpdateCloudReplicaInfo info) throws MetaNot } } + // 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)) { + 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) { @@ -1181,15 +1260,15 @@ private void unprotectUpdateCloudReplica(OlapTable olapTable, UpdateCloudReplica 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()); @@ -1205,9 +1284,9 @@ private void unprotectUpdateCloudReplica(OlapTable olapTable, UpdateCloudReplica } 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; @@ -1215,7 +1294,8 @@ private void unprotectUpdateCloudReplica(OlapTable olapTable, UpdateCloudReplica 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 05c4e04e7dee73..f53c4c0833a0ae 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 @@ -110,6 +110,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 availableBeIds, long idx) { return getCloudColocateHrwBeIdInternal(groupId, clusterId, availableBeIds, idx, -1); } @@ -985,6 +992,7 @@ public void replayDropBackend(Backend backend) { List 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 1fe717d59ff4d6..a73e14c59d968c 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.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; @@ -45,6 +46,7 @@ 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; @@ -160,6 +162,13 @@ public synchronized void doCheckpoint() throws CheckpointException { } 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(); @@ -410,6 +419,16 @@ public ReentrantReadWriteLock getLock() { 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 14f38a680e4338..f08becb81b0821 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 @@ -3093,12 +3093,7 @@ public TGetTabletReplicaInfosResult getTabletReplicaInfos(TGetTabletReplicaInfos // 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 00000000000000..d3f4a08943261d --- /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 com.google.common.collect.ImmutableMap; +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.concurrent.atomic.AtomicLong; + +public class CloudEnvRouteCleanupTest { + private boolean savedClean; + private CloudEnv env; + private CloudSystemInfoService systemInfo; + private CloudInternalCatalog catalog; + private MockedStatic 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(ImmutableMap.of(Tag.TYPE_LOCATION, "default", + Tag.CLOUD_CLUSTER_ID, "cg", Tag.CLOUD_CLUSTER_NAME, "cg")); + return backend; + } + + private static Pair 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 6a2614c39944d5..3bafb19bea2de6 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 @@ -25,6 +25,7 @@ import org.apache.doris.cloud.qe.ComputeGroupException; import org.apache.doris.cloud.system.CloudSystemInfoService; import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.system.Backend; @@ -354,6 +355,94 @@ private void stubBackend(long id, Backend 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(mockInfoService)); + + 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(mockInfoService)); + + 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(mockInfoService)); + + // 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) { 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 10cd5a9847cbc8..68374d2bde7adc 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 @@ -19,6 +19,7 @@ import org.apache.doris.cloud.system.CloudSystemInfoService; import org.apache.doris.common.Config; +import org.apache.doris.common.util.TimeUtils; import org.apache.doris.metric.MetricRepo; import org.junit.jupiter.api.AfterEach; @@ -31,8 +32,11 @@ import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.time.LocalTime; +import java.time.format.DateTimeFormatter; import java.util.AbstractMap; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; @@ -304,4 +308,144 @@ public void testResetCloudBalanceMetric_clearsMetricForAllClusters() throws Exce "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 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 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)))); + + // 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)); + Config.enable_cloud_replica_stale_route_clean = true; + Assertions.assertTrue(rebalancer.staleRouteSweepNeeded(bes)); + } finally { + 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 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 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 2f61d97f715633..bcca81f90cf0ed 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.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 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 testCreateTableConflictDropsOnlyLosingIndexes() throws Exception { testCreateTableConflictCleanup(Cloud.MetaServiceCode.OK); @@ -114,6 +166,138 @@ public void testCreateTabletMetaUsesCurrentSchemaVersionAndFormat() throws Excep } } + @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 replicas = Arrays.asList( + addReplayReplica(60001L, 70001L), addReplayReplica(60002L, 70002L)); + replicas.forEach(replica -> replica.updateClusterToPrimaryBe(STALE_CLUSTER_ID, DEAD_BE_ID)); + List tabletIds = Arrays.asList(60001L, 60002L); + List 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 00000000000000..019d8c86ad723c --- /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 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)) + .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)) + .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)) + .thenReturn(Collections.singletonList(materializedIndex)); + Mockito.when(partition.getMaterializedIndices(IndexExtState.ALL)) + .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 new file mode 100644 index 00000000000000..2239266814e11d --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplCloudTest.java @@ -0,0 +1,70 @@ +// 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.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.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.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +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 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; + } + } + +}