This is an automated email from the ASF dual-hosted git repository.
hello-stephen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a6ab604907e [fix](cloud) Exclude decommissioning BE from BE selection
(#65221)
a6ab604907e is described below
commit a6ab604907e752d71f1359d5de9b84a7ca170a77
Author: deardeng <[email protected]>
AuthorDate: Wed Jul 8 19:48:30 2026 +0800
[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.
### Release note
Cloud backend selection avoids routing new replica, routine load, and
Kafka proxy work to decommissioning backends.
---
.../apache/doris/cloud/catalog/CloudReplica.java | 37 +++++++--
.../doris/cloud/catalog/CloudReplicaTest.java | 91 +++++++++++++++++++++-
2 files changed, 121 insertions(+), 7 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 c779b8de053..c663efe1f8c 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
@@ -108,6 +108,21 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
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 {
List<Backend> clusterBackends = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
.getBackendsByClusterId(clusterId);
@@ -133,7 +148,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
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);
@@ -162,7 +177,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
.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();
}
}
@@ -296,6 +311,10 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
int indexRand = rand.nextInt(Config.cloud_replica_num);
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;
@@ -309,14 +328,14 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
// 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();
}
}
@@ -328,6 +347,12 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
// 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 {
@@ -386,7 +411,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
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);
@@ -452,7 +477,7 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
// 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);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
index 8ad3acf1fad..6edff8211c2 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
@@ -56,10 +56,16 @@ public class CloudReplicaTest {
private CloudSystemInfoService mockInfoService;
private int savedRehashSeconds;
+ private boolean savedEnableCloudMultiReplica;
+ private int savedCloudReplicaNum;
+ private String savedCloudUniqueId;
@BeforeEach
public void setUp() {
savedRehashSeconds = Config.rehash_tablet_after_be_dead_seconds;
+ savedEnableCloudMultiReplica = Config.enable_cloud_multi_replica;
+ savedCloudReplicaNum = Config.cloud_replica_num;
+ savedCloudUniqueId = Config.cloud_unique_id;
Config.cloud_unique_id = "test_cloud_unique_id";
mockEnv = Mockito.mock(Env.class);
@@ -76,6 +82,9 @@ public class CloudReplicaTest {
public void tearDown() {
envMockedStatic.close();
Config.rehash_tablet_after_be_dead_seconds = savedRehashSeconds;
+ Config.enable_cloud_multi_replica = savedEnableCloudMultiReplica;
+ Config.cloud_replica_num = savedCloudReplicaNum;
+ Config.cloud_unique_id = savedCloudUniqueId;
}
private CloudReplica createReplica() {
@@ -84,11 +93,16 @@ public class CloudReplicaTest {
}
private Backend createBackend(long id, boolean alive, boolean
decommissioned) {
+ return createBackend(id, alive, false, decommissioned);
+ }
+
+ private Backend createBackend(long id, boolean alive, boolean
decommissioning, boolean decommissioned) {
Backend be = Mockito.mock(Backend.class);
Mockito.when(be.getId()).thenReturn(id);
Mockito.when(be.isAlive()).thenReturn(alive);
+ Mockito.when(be.isDecommissioning()).thenReturn(decommissioning);
Mockito.when(be.isDecommissioned()).thenReturn(decommissioned);
- Mockito.when(be.isQueryAvailable()).thenReturn(alive &&
!decommissioned);
+ Mockito.when(be.isQueryAvailable()).thenReturn(alive);
Mockito.when(be.getLastUpdateMs()).thenReturn(System.currentTimeMillis());
return be;
}
@@ -234,6 +248,81 @@ public class CloudReplicaTest {
"Should fall back to decommissioned BE when no normal alive
BE");
}
+ @Test
+ public void
testGetColocatedBeId_decommissioningBeExcludedFromNormalHashRing() throws
ComputeGroupException {
+ setupColocateTable();
+
+ Backend be1 = createBackend(1001L, true, false);
+ Backend be2Decommissioning = createBackend(1002L, true, true, false);
+
+ Mockito.when(mockInfoService.getBackendsByClusterId(CLUSTER_ID_1))
+ .thenReturn(new ArrayList<>(Arrays.asList(be1,
be2Decommissioning)));
+
Mockito.when(mockInfoService.getClusterNameByClusterId(CLUSTER_ID_1)).thenReturn(CLUSTER_NAME_1);
+
+ CloudReplica replica = createReplica();
+ long pickedBeId = replica.getColocatedBeId(CLUSTER_ID_1);
+
+ Assertions.assertEquals(1001L, pickedBeId,
+ "Should exclude decommissioning BE from normal colocate hash
ring");
+ }
+
+ @Test
+ public void
testHashReplicaToBe_decommissioningBeExcludedFromNormalHashRing() throws
ComputeGroupException {
+ Backend be1 = createBackend(1001L, true, false);
+ Backend be2Decommissioning = createBackend(1002L, true, true, false);
+
+ Mockito.when(mockInfoService.getBackendsByClusterId(CLUSTER_ID_1))
+ .thenReturn(new ArrayList<>(Arrays.asList(be1,
be2Decommissioning)));
+
Mockito.when(mockInfoService.getClusterNameByClusterId(CLUSTER_ID_1)).thenReturn(CLUSTER_NAME_1);
+
+ CloudReplica replica = createReplica();
+ long pickedBeId = replica.hashReplicaToBe(CLUSTER_ID_1, false);
+
+ Assertions.assertEquals(1001L, pickedBeId,
+ "Should exclude decommissioning BE from normal replica hash
ring");
+ }
+
+ @Test
+ public void
testGetBackendId_multiReplicaDecommissioningBeExcludedFromNormalHashRing()
+ throws ComputeGroupException {
+
Mockito.when(mockColocateIndex.isColocateTableNoLock(TABLE_ID)).thenReturn(false);
+ Config.enable_cloud_multi_replica = true;
+ Config.cloud_replica_num = 2;
+
+ Backend be1 = createBackend(1001L, true, false);
+ Backend be2Decommissioning = createBackend(1002L, true, true, false);
+
+ Mockito.when(mockInfoService.getBackendsByClusterId(CLUSTER_ID_1))
+ .thenReturn(new ArrayList<>(Arrays.asList(be1,
be2Decommissioning)));
+
Mockito.when(mockInfoService.getClusterNameByClusterId(CLUSTER_ID_1)).thenReturn(CLUSTER_NAME_1);
+
+ CloudReplica replica = createReplica();
+ long pickedBeId = replica.getBackendIdWithClusterId(CLUSTER_ID_1);
+
+ Assertions.assertEquals(1001L, pickedBeId,
+ "Should exclude decommissioning BE from normal multi-replica
hash ring");
+ }
+
+ @Test
+ public void testGetBackendId_decommissioningPrimaryBeTriggersRehash()
throws ComputeGroupException {
+
Mockito.when(mockColocateIndex.isColocateTableNoLock(TABLE_ID)).thenReturn(false);
+
+ Backend be1 = createBackend(1001L, true, false);
+ Backend be2Decommissioning = createBackend(1002L, true, true, false);
+
Mockito.when(mockInfoService.getBackend(1002L)).thenReturn(be2Decommissioning);
+ Mockito.when(mockInfoService.getBackendsByClusterId(CLUSTER_ID_1))
+ .thenReturn(new ArrayList<>(Arrays.asList(be1,
be2Decommissioning)));
+
Mockito.when(mockInfoService.getClusterNameByClusterId(CLUSTER_ID_1)).thenReturn(CLUSTER_NAME_1);
+
+ CloudReplica replica = createReplica();
+ replica.updateClusterToPrimaryBe(CLUSTER_ID_1, 1002L);
+
+ long pickedBeId = replica.getBackendIdWithClusterId(CLUSTER_ID_1);
+
+ Assertions.assertEquals(1001L, pickedBeId,
+ "Should rehash instead of returning cached decommissioning
primary BE");
+ }
+
// ---------------------------------------------------------------
// Tests for getAllPrimaryBes: colocate table support
// ---------------------------------------------------------------
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
