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 1e109c3350e74a..29a1cf9cd71fac 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 @@ -114,6 +114,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() @@ -129,7 +144,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); @@ -158,7 +173,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(); } } @@ -279,6 +294,10 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException { } 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; @@ -292,14 +311,14 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException { // use primaryClusterToBackends, 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(); } } @@ -311,6 +330,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 { @@ -369,7 +394,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); @@ -435,7 +460,7 @@ public List hashReplicaToBes(String clusterId, boolean isBackGround, int r // 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);