From 9a053d7edf3418188c33c86f50f774bdfc2f7d49 Mon Sep 17 00:00:00 2001 From: deardeng Date: Wed, 8 Jul 2026 19:48:30 +0800 Subject: [PATCH] [fix](cloud) Exclude decommissioning BE from BE selection (#65221) Problem Summary: Cloud decommissioning backends can still be alive before they reach the final decommissioned state. These backends could still be selected by CloudReplica routing, routine load assignment, and Kafka proxy backend selection. This change treats decommissioning backends the same as decommissioned backends in those selection paths, while preserving current master behavior such as direct multi-replica hashing and routine load blacklist stale-backend cleanup. Cloud backend selection avoids routing new replica, routine load, and Kafka proxy work to decommissioning backends. (cherry picked from commit a6ab604907e752d71f1359d5de9b84a7ca170a77) --- .../doris/cloud/catalog/CloudReplica.java | 37 ++++++++++++++++--- 1 file changed, 31 insertions(+), 6 deletions(-) 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 07a2f4166354d6..04b4ca167db500 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 @@ -106,6 +106,21 @@ private boolean isColocated() { return Env.getCurrentColocateIndex().isColocateTableNoLock(tableId); } + private boolean isDecommissioningOrDecommissioned(Backend be) { + boolean decommissioning = be.isDecommissioning(); + boolean decommissioned = be.isDecommissioned(); + if ((decommissioning || decommissioned) && LOG.isDebugEnabled()) { + LOG.debug("backend {} is filtered by decommission state, decommissioning={}, decommissioned={}, " + + "replica info {}", + be.getId(), decommissioning, decommissioned, this); + } + return decommissioning || decommissioned; + } + + private boolean isQueryAvailableAndNotDecommissioning(Backend be) { + return be != null && be.isQueryAvailable() && !isDecommissioningOrDecommissioned(be); + } + public long getColocatedBeId(String clusterId) throws ComputeGroupException { CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo()); List bes = infoService.getBackendsByClusterId(clusterId).stream() @@ -121,7 +136,7 @@ public long getColocatedBeId(String clusterId) throws ComputeGroupException { List decommissionAvailBes = new ArrayList<>(); for (Backend be : bes) { if (be.isAlive()) { - if (be.isDecommissioned()) { + if (isDecommissioningOrDecommissioned(be)) { decommissionAvailBes.add(be); } else { availableBes.add(be); @@ -150,7 +165,7 @@ public long getColocatedBeId(String clusterId) throws ComputeGroupException { .collect(Collectors.toList()); long index = getIndexByBeNum(hashCode.asLong() + idx, beAliveOrDeadShort.size()); Backend be = beAliveOrDeadShort.get((int) index); - if (be.isAlive() && !be.isDecommissioned()) { + if (be.isAlive() && !isDecommissioningOrDecommissioned(be)) { return be.getId(); } } @@ -243,6 +258,10 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException { int indexRand = rand.nextInt(Config.cloud_replica_num); List res = hashReplicaToBes(clusterId, false, Config.cloud_replica_num); + if (LOG.isDebugEnabled()) { + LOG.debug("rehash multi replica backend, clusterId {}, replica info {}, indexRand {}, hashedBes {}", + clusterId, this, indexRand, res); + } if (res.size() < indexRand + 1) { if (res.isEmpty()) { return -1; @@ -256,14 +275,14 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException { // use primaryClusterToBackend, if find be normal Backend be = getPrimaryBackend(clusterId, false); - if (be != null && be.isQueryAvailable()) { + if (isQueryAvailableAndNotDecommissioning(be)) { return be.getId(); } if (!Config.enable_immediate_be_assign) { // use secondaryClusterToBackends, if find be normal be = getSecondaryBackend(clusterId); - if (be != null && be.isQueryAvailable()) { + if (isQueryAvailableAndNotDecommissioning(be)) { return be.getId(); } } @@ -275,6 +294,12 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException { // be abnormal, rehash it. configure settings to different maps long pickBeId = hashReplicaToBe(clusterId, false); + if (LOG.isDebugEnabled()) { + LOG.debug("rehash replica backend, clusterId {}, pickedBeId {}, immediateAssign {}, replica info {}, " + + "primaryBackend {}, secondaryBackend {}", + clusterId, pickBeId, Config.enable_immediate_be_assign, this, getPrimaryBackend(clusterId, false), + getSecondaryBackend(clusterId)); + } if (Config.enable_immediate_be_assign) { updateClusterToPrimaryBe(clusterId, pickBeId); } else { @@ -333,7 +358,7 @@ public long hashReplicaToBe(String clusterId, boolean isBackGround) throws Compu List decommissionAvailBes = new ArrayList<>(); for (Backend be : clusterBes) { if (be.isQueryAvailable() && !be.isSmoothUpgradeSrc()) { - if (be.isDecommissioned()) { + if (isDecommissioningOrDecommissioned(be)) { decommissionAvailBes.add(be); } else { availableBes.add(be); @@ -399,7 +424,7 @@ private List hashReplicaToBes(String clusterId, boolean isBackGround, int // be core or restart must in heartbeat_interval_second if ((be.isAlive() || missTimeMs <= Config.heartbeat_interval_second * 1000L) && !be.isSmoothUpgradeSrc()) { - if (be.isDecommissioned()) { + if (isDecommissioningOrDecommissioned(be)) { decommissionAvailBes.add(be); } else { availableBes.add(be);