yiguolei · Oct 23, 2026 · Oct 14, 2026
@@ -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. "

@@ -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();

@@ -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,26 +670,31 @@ 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<String, Long> 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;
     }
 
     /**
      * Returns the set of compute group IDs that have primary backends for this replica.
-     * Used by lazy fetch path to also collect secondary backends per compute group.
      */
     public Set<String> getPrimaryComputeGroupIds() {
         return primaryClusterToBackend.keySet();
@@ -707,6 +719,32 @@ public void checkAndClearSecondaryClusterToBe(String clusterId, long expireTimes
         }
     }
 
+    public List<Backend> getAllSecondaryBes() {
+        List<Backend> result = new ArrayList<>();
+        SystemInfoService systemInfo = Env.getCurrentSystemInfo();
+        for (Pair<Long, Long> route : secondaryClusterToBackends.values()) {
+            Backend backend = systemInfo.getBackendByIdWithBoxedId(route.first);
+            if (backend != null) {
+                result.add(backend);
+            }
+        }
+        return result;
+    }
+
+    /**
+     * @param systemInfo the backend set to judge staleness against -- see {@link #removeInvalidRoutes} for why
+     *                    this must come from the caller's Env rather than Env.getCurrentSystemInfo().
+     */
+    public void replayUpdateClusterToPrimaryBe(String clusterId, long beId, SystemInfoService systemInfo) {
+        // A journal queued before backend deletion can arrive after the deletion was replayed.
+        // Do not publish that route, or clear a live secondary while rejecting it.
+        if (Config.enable_cloud_replica_stale_route_clean && !FeConstants.runningUnitTest
+                && systemInfo.getBackend(beId) == null) {
+            return;
+        }
+        updateClusterToPrimaryBe(clusterId, beId);
+    }
+
     public List<Backend> getAllPrimaryBes() {
         List<Backend> result = new ArrayList<Backend>();
         if (isColocated()) {
@@ -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<String, Long> sharedRoutes = new ConcurrentHashMap<>(primaryClusterToBackend.size());
         primaryClusterToBackend.forEach((clusterId, backendId) ->

@@ -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<Long, Set<Long>> DEFAULT_GLOBAL_TABLET_SET_FACTORY =
             ignored -> ConcurrentHashMap.newKeySet();
+    private static final DateTimeFormatter STALE_ROUTE_CLEAN_TIME_FORMAT = DateTimeFormatter.ofPattern("HH:mm");
 
     private final CloudTabletRebalancerMetrics rebalancerMetrics;
     private long currentRoundTabletScanCount;
@@ -1005,24 +1011,62 @@ boolean staleRouteSweepNeeded(Set<Long> 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<UpdateCloudReplicaInfo> updateReplicaInfos = new ArrayList<UpdateCloudReplicaInfo>();
         long[] assignedErrNum = {0L};
-        long[] staleRouteNum = {0L};
         boolean sweepStaleRoutes = staleRouteSweepNeeded(allBes);
-        // loopCloudReplica() has the compute group loop innermost, so it hands us every replica once per
-        // live compute group, while removeInvalidRoutes() scans the whole route map and does not care
-        // which group we are on. Pin the sweep to one arbitrary group id so a sweeping round still makes a
-        // single pass per replica. If clusterToBes is empty the callback never runs at all, so the serving
-        // catalog keeps the entries until it reloads the image -- there is nothing to route in that state.
-        String sweepTicket = sweepStaleRoutes ? clusterToBes.keySet().stream().findFirst().orElse(null) : null;
+        // Cleanup is independent of compute groups and includes shadow indices of active tables.
+        // Take the catalog and backend set from the same Env so the sweep can never mix the two.
+        Env currentEnv = Env.getCurrentEnv();
+        long staleRouteNum = sweepStaleRoutes
+                ? ((CloudInternalCatalog) currentEnv.getInternalCatalog())
+                        .removeInvalidCloudReplicaRoutes(currentEnv.getClusterInfo())
+                : 0;
         long needRehashDeadTime = System.currentTimeMillis() - Config.rehash_tablet_after_be_dead_seconds * 1000L;
         loopCloudReplica((Database db, Table table, Partition partition, MaterializedIndex index, String cluster) -> {
             boolean assigned = false;
@@ -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;