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; + } + } + +}