Skip to content

[fix](cloud) Exclude decommissioning BE from BE selection (#65221) #66070

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deardeng wants to merge 1 commit into
base: branch-4.0
Choose a base branch
from
+31 −6
Open

[fix](cloud) Exclude decommissioning BE from BE selection (#65221) #66070

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 31 additions & 6 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Backend> bes = infoService.getBackendsByClusterId(clusterId).stream()
Expand All @@ -129,7 +144,7 @@ public long getColocatedBeId(String clusterId) throws ComputeGroupException {
List<Backend> decommissionAvailBes = new ArrayList<>();
for (Backend be : bes) {
if (be.isAlive()) {
if (be.isDecommissioned()) {
if (isDecommissioningOrDecommissioned(be)) {
decommissionAvailBes.add(be);
} else {
availableBes.add(be);
Expand Down Expand Up @@ -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();
}
}
Expand Down Expand Up @@ -279,6 +294,10 @@ private long getBackendIdImpl(String clusterId) throws ComputeGroupException {
}

List<Long> 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;
Expand All @@ -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();
}
}
Expand All @@ -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 {
Expand Down Expand Up @@ -369,7 +394,7 @@ public long hashReplicaToBe(String clusterId, boolean isBackGround) throws Compu
List<Backend> 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);
Expand Down Expand Up @@ -435,7 +460,7 @@ public List<Long> 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);
Expand Down
Loading