[fix](cloud) Drop CloudReplica route entries of dropped compute groups (#66984)

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3807,6 +3807,13 @@ public static int metaServiceRpcRetryTimes() {
             description = { "存算分离模式下，一个 BE 挂掉多长时间后，它的 tablet 彻底转移到其他 BE 上" })
     public static int rehash_tablet_after_be_dead_seconds = 3600;
 
+    @ConfField(mutable = true, masterOnly = false,
+            description = "Whether to drop the primary/secondary route entries of a CloudReplica whose backend no "
+                    + "longer exists, when loading the image and in the tablet rebalancer round. Those entries are "
+                    + "already ignored at query time (the replica is rehashed), so they only waste FE memory and "
+                    + "image size. Set to false to keep the legacy leaking behavior. Default is true.")
+    public static boolean enable_cloud_replica_stale_route_clean = true;
+
     @ConfField(mutable = false, masterOnly = true,
             description = {
                     "Whether to use rendezvous hashing for colocate bucket placement in cloud mode. "

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java

@@ -23,6 +23,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.common.Pair;
 import org.apache.doris.common.util.DebugPointUtil;
 import org.apache.doris.persist.gson.GsonPostProcessable;
@@ -385,9 +386,34 @@ clusterId, pickBeId, Config.enable_immediate_be_assign, this, getPrimaryBackend(
         } else {
             updateClusterToSecondaryBe(clusterId, pickBeId);
         }
+        discardRouteIfBackendVanished(pickBeId);
         return pickBeId;
     }
 
+    /**
+     * The compute group can be dropped while a publisher is picking a backend out of it, in which case the
+     * route just written would outlive the group. Re-checking right after publishing closes that window:
+     * either the backend is already gone and the write is undone here, or it was still registered, which
+     * means the drop -- and the rebalancer sweep it triggers -- happens after this write and will visit
+     * this replica. Ordering, not timing, is what makes the sweep sufficient; a publisher may stall for
+     * arbitrarily long between choosing a backend and publishing without escaping cleanup.
+     *
+     * Every route publisher that resolves a backend outside the rebalancer round must call this, and must
+     * not persist the route when it returns false.
+     *
+     * @return false when the route was discarded because its backend is gone
+     */
+    public boolean discardRouteIfBackendVanished(long beId) {
+        if (!Config.enable_cloud_replica_stale_route_clean) {
+            return true;
+        }
+        if (Env.getCurrentSystemInfo().getBackend(beId) == null) {
+            removeInvalidRoutes();
+            return false;
+        }
+        return true;
+    }
+
     public Backend getPrimaryBackend(String clusterId, boolean setIfAbsent) {
         long beId = getClusterPrimaryBackendId(clusterId);
         if (beId != -1L) {
@@ -399,6 +425,7 @@ public Backend getPrimaryBackend(String clusterId, boolean setIfAbsent) {
                 try {
                     beId = getBackendIdImpl(clusterId);
                     updateClusterToPrimaryBe(clusterId, beId);
+                    discardRouteIfBackendVanished(beId);
                     return Env.getCurrentSystemInfo().getBackend(beId);
                 } catch (ComputeGroupException e) {
                     return null;
@@ -593,6 +620,48 @@ public void clearClusterToBe(String cluster) {
         secondaryClusterToBackends.remove(cluster);
     }
 
+    /**
+     * Drop the route entries whose backend has been dropped from the cluster.
+     *
+     * Such an entry is already dead weight: getBackendIdImpl() resolves the backend id, gets null and
+     * falls back to hashReplicaToBe(), so removing it does not change routing. But nothing ever removes
+     * it either -- dropCluster() only touches CloudSystemInfoService, and the rebalancer only walks the
+     * compute groups that currently exist -- so entries of dropped compute groups pile up forever, both
+     * in FE heap and in the image (the `bes`/`be` field).
+     *
+     * @return how many entries were dropped
+     */
+    public int removeInvalidRoutes() {
+        if (!Config.enable_cloud_replica_stale_route_clean || FeConstants.runningUnitTest) {
+            return 0;
+        }
+        SystemInfoService systemInfo = Env.getCurrentSystemInfo();
+        // Remove conditionally rather than by predicate: route writers run concurrently, so comparing map
+        // sizes before and after would mix their insertions into the count (and could even report a
+        // negative one), and a key whose value was just rewritten to a live backend must not be dropped.
+        int removed = 0;
+        if (!secondaryClusterToBackends.isEmpty()) {
+            for (Map.Entry<String, Pair<Long, Long>> entry : secondaryClusterToBackends.entrySet()) {
+                if (systemInfo.getBackend(entry.getValue().key()) == null
+                        && secondaryClusterToBackends.remove(entry.getKey(), entry.getValue())) {
+                    removed++;
+                }
+            }
+        }
+        // Keep a dead primary whose compute group still has a live secondary. With
+        // enable_immediate_be_assign=false that is the normal failover state, and the lazy fetch path in
+        // FrontendServiceImpl.getTabletReplicaInfos() enumerates secondaries through the primary key set,
+        // so dropping the key would hide a live secondary from the peer cache candidates.
+        for (Map.Entry<String, Long> entry : primaryClusterToBackend.entrySet()) {
+            if (systemInfo.getBackend(entry.getValue()) == null
+                    && !secondaryClusterToBackends.containsKey(entry.getKey())
+                    && primaryClusterToBackend.remove(entry.getKey(), entry.getValue())) {
+                removed++;
+            }
+        }
+        return removed;
+    }
+
     /**
      * Returns the set of compute group IDs that have primary backends for this replica.
      * Used by lazy fetch path to also collect secondary backends per compute group.
@@ -666,5 +735,10 @@ public void gsonPostProcess() throws IOException {
             }
             this.primaryClusterToBackends = null;
         }
+        // outside the `bes` branch on purpose: the new `be` format accumulates stale entries just the same.
+        // The backends module is loaded before db/recycleBin (PersistMetaModules.MODULE_NAMES), and the
+        // checkpoint thread resolves Env.getCurrentEnv() to its own Env, so the backend set read here is
+        // the one belonging to the image being loaded.
+        removeInvalidRoutes();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTabletRebalancer.java

@@ -52,6 +52,7 @@
 import org.apache.doris.thrift.TWarmUpCacheAsyncRequest;
 import org.apache.doris.thrift.TWarmUpCacheAsyncResponse;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Sets;
@@ -97,6 +98,9 @@ public class CloudTabletRebalancer extends MasterDaemon {
     private Map<String, List<Long>> clusterToBes;
 
     private Set<Long> allBes;
+    // backend baseline and remaining sweep rounds, see staleRouteSweepNeeded()
+    private Set<Long> lastSweptBackends = null;
+    private int pendingSweepRounds = 0;
 
     // partitionId -> indexId -> be -> tabletIds
     private ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Long>>>> partitionToTablets;
@@ -933,9 +937,47 @@ public void checkDecommissionState(Map<String, List<Long>> clusterToBes) {
         }
     }
 
+    /**
+     * Decides whether this round sweeps stale routes, and advances the backend baseline. Call once per
+     * round. Without this gate the sweep would walk every replica's route maps once a second under
+     * table.readLock() only to find nothing, which on a large catalog is pure allocation.
+     */
+    @VisibleForTesting
+    boolean staleRouteSweepNeeded(Set<Long> currentBes) {
+        if (!Config.enable_cloud_replica_stale_route_clean) {
+            lastSweptBackends = null;
+            pendingSweepRounds = 0;
+            return false;
+        }
+        if (lastSweptBackends == null || !currentBes.containsAll(lastSweptBackends)) {
+            // Only a backend that went away can strand a route. Two rounds rather than one: a query
+            // thread can pick a backend in hashReplicaToBe() before the drop and publish the route in
+            // getBackendIdImpl() after this pass already visited that replica, so the extra round
+            // catches writers that were in flight during the first one.
+            pendingSweepRounds = 2;
+        }
+        // An addition strands nothing, so it does not trigger a sweep -- but it must still enter the
+        // baseline. Advancing only after a sweep would leave the baseline at the pre-addition set, and
+        // dropping that same backend later would compare equal to it and go unnoticed.
+        lastSweptBackends = currentBes;
+        if (pendingSweepRounds > 0) {
+            pendingSweepRounds--;
+            return true;
+        }
+        return false;
+    }
+
     private boolean completeRouteInfo() {
         List<UpdateCloudReplicaInfo> updateReplicaInfos = new ArrayList<UpdateCloudReplicaInfo>();
         long[] assignedErrNum = {0L};
+        long[] staleRouteNum = {0L};
+        boolean sweepStaleRoutes = staleRouteSweepNeeded(allBes);
+        // loopCloudReplica() has the compute group loop innermost, so it hands us every replica once per
+        // live compute group, while removeInvalidRoutes() scans the whole route map and does not care
+        // which group we are on. Pin the sweep to one arbitrary group id so a sweeping round still makes a
+        // single pass per replica. If clusterToBes is empty the callback never runs at all, so the serving
+        // catalog keeps the entries until it reloads the image -- there is nothing to route in that state.
+        String sweepTicket = sweepStaleRoutes ? clusterToBes.keySet().stream().findFirst().orElse(null) : null;
         long needRehashDeadTime = System.currentTimeMillis() - Config.rehash_tablet_after_be_dead_seconds * 1000L;
         loopCloudReplica((Database db, Table table, Partition partition, MaterializedIndex index, String cluster) -> {
             boolean assigned = false;
@@ -945,6 +987,16 @@ private boolean completeRouteInfo() {
             for (Tablet tablet : index.getTablets()) {
                 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);
                     InfightTablet taskKey = new InfightTablet(tablet.getId(), cluster);
@@ -1011,7 +1063,8 @@ private boolean completeRouteInfo() {
             }
         });
 
-        LOG.info("collect to editlog route {} infos, error num {}", updateReplicaInfos.size(), assignedErrNum[0]);
+        LOG.info("collect to editlog route {} infos, error num {}, swept stale routes {}, entries dropped {}",
+                updateReplicaInfos.size(), assignedErrNum[0], sweepStaleRoutes, staleRouteNum[0]);
 
         if (updateReplicaInfos.isEmpty()) {
             return true;
@@ -1678,6 +1731,14 @@ private void updateClusterToBeMap(long tabletId, long destBe, String clusterId,
             }
 
             cloudReplica.updateClusterToPrimaryBe(clusterId, destBe);
+            if (!cloudReplica.discardRouteIfBackendVanished(destBe)) {
+                // The warmup checker resolves its destination on its own scheduler thread and can stall
+                // there while the compute group is dropped and the sweeps triggered by that drop finish.
+                // Journalling the route now would hand every FE an entry nothing is left to clean.
+                LOG.info("compute group {} lost backend {} while warming up tablet {}, dropping the route",
+                        clusterId, destBe, tabletId);
+                return;
+            }
             UpdateCloudReplicaInfo info = new UpdateCloudReplicaInfo(tabletMeta.getDbId(),
                     tabletMeta.getTableId(), tabletMeta.getPartitionId(), tabletMeta.getIndexId(),
                     tabletId, cloudReplica.getId(), clusterId, destBe);