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

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

@@ -3661,6 +3661,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;
@@ -410,9 +411,24 @@ clusterId, pickBeId, Config.enable_immediate_be_assign, this, getPrimaryBackend(
         } else {
             updateClusterToSecondaryBe(clusterId, pickBeId);
         }
+        discardRouteIfBackendVanished(pickBeId);
         return pickBeId;
     }
 
+    /**
+     * Discard a route if its selected backend disappeared while it was being published.
+     */
+    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) {
@@ -424,6 +440,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;
@@ -629,6 +646,35 @@ public void clearClusterToBe(String cluster) {
         secondaryClusterToBackends.remove(cluster);
     }
 
+    /**
+     * Drop route entries whose backend no longer exists.
+     *
+     * @return number of removed entries
+     */
+    public int removeInvalidRoutes() {
+        if (!Config.enable_cloud_replica_stale_route_clean || FeConstants.runningUnitTest) {
+            return 0;
+        }
+        SystemInfoService systemInfo = Env.getCurrentSystemInfo();
+        int removed = 0;
+        for (Map.Entry<String, Pair<Long, Long>> entry : secondaryClusterToBackends.entrySet()) {
+            if (systemInfo.getBackend(entry.getValue().key()) == null
+                    && secondaryClusterToBackends.remove(entry.getKey(), entry.getValue())) {
+                removed++;
+            }
+        }
+        for (Map.Entry<String, List<Long>> entry : primaryClusterToBackends.entrySet()) {
+            List<Long> backendIds = entry.getValue();
+            if (backendIds != null && !backendIds.isEmpty()
+                    && systemInfo.getBackend(backendIds.get(0)) == null
+                    && !secondaryClusterToBackends.containsKey(entry.getKey())
+                    && primaryClusterToBackends.remove(entry.getKey(), backendIds)) {
+                removed++;
+            }
+        }
+        return removed;
+    }
+
     // ATTN: This func is only used by redundant tablet report clean in bes.
     // Only the master node will do the diff logic,
     // so just only need to clean up secondaryClusterToBackends on the master node.
@@ -676,5 +722,6 @@ public void gsonPostProcess() throws IOException {
             }
             this.primaryClusterToBackend = null;
         }
+        removeInvalidRoutes();
     }
 }

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

@@ -54,6 +54,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Sets;
+import com.google.common.annotations.VisibleForTesting;
 import lombok.Getter;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -96,6 +97,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 -> tablet
     private ConcurrentHashMap<Long, ConcurrentHashMap<Long, ConcurrentHashMap<Long, Set<Tablet>>>> partitionToTablets;
@@ -938,9 +942,33 @@ public void checkDecommissionState(Map<String, List<Long>> clusterToBes) {
         }
     }
 
+    /**
+     * Decide whether this round should sweep stale route entries.
+     */
+    @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)) {
+            pendingSweepRounds = 2;
+        }
+        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);
+        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;
@@ -950,6 +978,9 @@ private boolean completeRouteInfo() {
             for (Tablet tablet : index.getTablets()) {
                 for (Replica r : tablet.getReplicas()) {
                     CloudReplica replica = (CloudReplica) r;
+                    if (cluster.equals(sweepTicket)) {
+                        staleRouteNum[0] += replica.removeInvalidRoutes();
+                    }
                     // clean secondary map
                     replica.checkAndClearSecondaryClusterToBe(cluster, needRehashDeadTime);
                     InfightTablet taskKey = new InfightTablet(tablet.getId(), cluster);
@@ -1016,7 +1047,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;
@@ -1633,6 +1665,11 @@ private void updateClusterToBeMap(Tablet pickedTablet, long destBe, String clust
             }
 
             cloudReplica.updateClusterToPrimaryBe(clusterId, destBe);
+            if (!cloudReplica.discardRouteIfBackendVanished(destBe)) {
+                LOG.info("compute group {} lost backend {} while warming up tablet {}, dropping the route",
+                        clusterId, destBe, pickedTablet.getId());
+                return;
+            }
             UpdateCloudReplicaInfo info = new UpdateCloudReplicaInfo(cloudReplica.getDbId(),
                     cloudReplica.getTableId(), cloudReplica.getPartitionId(), cloudReplica.getIndexId(),
                     pickedTablet.getId(), cloudReplica.getId(), clusterId, destBe);