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 80d2f9ba0b7aad..38d3b806019d8a 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 @@ -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. " 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 e6772c3c39d9f8..54e98aaf86a258 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 @@ -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> entry : secondaryClusterToBackends.entrySet()) { + if (systemInfo.getBackend(entry.getValue().key()) == null + && secondaryClusterToBackends.remove(entry.getKey(), entry.getValue())) { + removed++; + } + } + for (Map.Entry> entry : primaryClusterToBackends.entrySet()) { + List 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(); } } 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 69df53fcda94e1..b382a152dbc524 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 @@ -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> clusterToBes; private Set allBes; + // backend baseline and remaining sweep rounds, see staleRouteSweepNeeded() + private Set lastSweptBackends = null; + private int pendingSweepRounds = 0; // partitionId -> indexId -> be -> tablet private ConcurrentHashMap>>> partitionToTablets; @@ -938,9 +942,33 @@ public void checkDecommissionState(Map> clusterToBes) { } } + /** + * Decide whether this round should sweep stale route entries. + */ + @VisibleForTesting + boolean staleRouteSweepNeeded(Set 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 updateReplicaInfos = new ArrayList(); 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);