This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 e035164c7e5 [fix](group commit) Fix group commit routing for virtual
compute groups (#66585)
e035164c7e5 is described below
commit e035164c7e5735662975a4343e6c8e5828e5f1ce
Author: Jamie <[email protected]>
AuthorDate: Fri Aug 21 15:24:56 2026 +0800
[fix](group commit) Fix group commit routing for virtual compute groups
(#66585)
Related PR: #61555
Problem Summary:
`CloudSystemInfoService` already resolves a requested cloud cluster or
virtual compute group to its current active backend pool. Group commit
then incorrectly reused the requested cluster name when validating both
cached and random backends. For a VCG, that compared the logical VCG
name with each backend's physical compute group name, rejected every
healthy candidate, and ended with `No suitable backend`.
This change keeps VCG routing inside `CloudSystemInfoService`:
- Group commit continues to use the requested cluster as its cache
scope.
- A cached cloud backend is valid only when its ID is still present in
the current active backend pool and it remains load-available.
- Cache hits use a scoped backend lookup in `CloudSystemInfoService`;
the service resolves physical or virtual compute groups internally and
searches the authoritative backend list without materializing the full
pool.
- Random selection trusts the active backend pool supplied by
`CloudSystemInfoService` and applies only backend health and
decommission checks.
This preserves the stale-cache protection needed after topology changes
without requiring Group Commit to understand logical and physical
compute group names. If the active compute group changes, the old cached
backend is absent from the new active pool and random selection falls
back to an eligible backend.
The existing VCG Docker suite enables synchronous group commit before
and after failover and exercises both the master HTTP path and
follower-to-master forwarding path.
### Release note
Fix group commit stream loads through virtual compute groups.
---
.../doris/cloud/system/CloudSystemInfoService.java | 41 +++++
.../org/apache/doris/load/GroupCommitManager.java | 36 ++--
.../cloud/system/CloudSystemInfoServiceTest.java | 8 +
.../GroupCommitManagerBackendSelectionTest.java | 3 +-
.../apache/doris/load/GroupCommitManagerTest.java | 188 +++++++++++++++++++++
.../use_vcg_read_write.groovy | 15 +-
6 files changed, 271 insertions(+), 20 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index a2f5c5c5100..b99e9967800 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -1394,6 +1394,47 @@ public class CloudSystemInfoService extends
SystemInfoService {
}
}
+ /**
+ * Returns the backend only when it belongs to the backend pool currently
selected for
+ * {@code clusterName}. The name may identify a physical or virtual
compute group; virtual
+ * routing is resolved internally. Backend health and load availability
are not checked here.
+ *
+ * @return the matching backend, or null if the cluster or backend is
absent or the backend is
+ * outside the current pool
+ */
+ public Backend getBackendInCurrentCluster(String clusterName, long
backendId) {
+ rlock.lock();
+ try {
+ String physicalClusterName = getPhysicalCluster(clusterName);
+ String clusterId = clusterNameToId.get(physicalClusterName);
+ if (Strings.isNullOrEmpty(clusterId)) {
+ return null;
+ }
+ List<Backend> backends = clusterIdToBackend.get(clusterId);
+ if (backends == null) {
+ return null;
+ }
+
+ int low = 0;
+ int high = backends.size() - 1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ Backend backend = backends.get(mid);
+ int result = Long.compare(backend.getId(), backendId);
+ if (result < 0) {
+ low = mid + 1;
+ } else if (result > 0) {
+ high = mid - 1;
+ } else {
+ return backend;
+ }
+ }
+ return null;
+ } finally {
+ rlock.unlock();
+ }
+ }
+
public ImmutableMap<Long, Backend> getCloudIdToBackend(String clusterName)
{
rlock.lock();
try {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
b/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
index 6cca0d9fdcd..0da766fadda 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/GroupCommitManager.java
@@ -322,7 +322,7 @@ public class GroupCommitManager {
throw new LoadException("No alive backend");
}
// If the cached backend is not active or decommissioned, select a
random new backend.
- Long randomBackendId = getRandomCloudBackend(cacheKey, cluster,
tableId, backends);
+ Long randomBackendId = getRandomCloudBackend(cacheKey, tableId,
backends);
if (randomBackendId != null) {
return randomBackendId;
}
@@ -397,8 +397,16 @@ public class GroupCommitManager {
if (pressure == null) {
return null;
} else if (pressure.get() < table.getGroupCommitDataBytes()) {
- Backend backend =
Env.getCurrentSystemInfo().getBackend(backendId);
- if (isBackendAvailable(backend, cloudCluster)) {
+ Backend backend;
+ if (cloudCluster != null) {
+ // The cloud service resolves a cluster or VCG to its
current active backend pool.
+ // Look up the cached backend in that pool to validate
membership after topology changes.
+ backend = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
+ .getBackendInCurrentCluster(cloudCluster,
backendId);
+ } else {
+ backend = Env.getCurrentSystemInfo().getBackend(backendId);
+ }
+ if (isBackendAvailable(backend)) {
return backend.getId();
} else {
tableToBeMap.invalidate(cacheKey);
@@ -410,23 +418,17 @@ public class GroupCommitManager {
return null;
}
- private boolean isBackendAvailable(Backend backend, @Nullable String
cloudCluster) {
- if (backend == null || !backend.isAlive() ||
backend.isDecommissioned() || backend.isDecommissioning()
- || !backend.isLoadAvailable()) {
- return false;
- }
- if (!Config.isCloudMode()) {
- return true;
- }
- return cloudCluster == null ||
cloudCluster.equals(backend.getCloudClusterName());
+ private boolean isBackendAvailable(Backend backend) {
+ return backend != null && backend.isAlive() &&
!backend.isDecommissioned()
+ && !backend.isDecommissioning() && backend.isLoadAvailable();
}
@Nullable
- private Long getRandomCloudBackend(String cacheKey, String cluster, long
tableId, List<Backend> backends)
+ private Long getRandomCloudBackend(String cacheKey, long tableId,
List<Backend> backends)
throws LoadException {
OlapTable table = (OlapTable)
Env.getCurrentEnv().getInternalCatalog().getTableByTableId(tableId);
Collections.shuffle(backends);
- return selectAvailableBackend(cacheKey, cluster, tableId, table,
backends);
+ return selectAvailableBackend(cacheKey, tableId, table, backends);
}
@Nullable
@@ -443,14 +445,14 @@ public class GroupCommitManager {
} catch (UserException e) {
throw new LoadException(e.getMessage());
}
- return selectAvailableBackend(cacheKey, null, tableId, table,
orderedBackends);
+ return selectAvailableBackend(cacheKey, tableId, table,
orderedBackends);
}
@Nullable
- private Long selectAvailableBackend(String cacheKey, @Nullable String
cloudCluster, long tableId, OlapTable table,
+ private Long selectAvailableBackend(String cacheKey, long tableId,
OlapTable table,
List<Backend> orderedBackends) {
for (Backend backend : orderedBackends) {
- if (isBackendAvailable(backend, cloudCluster)) {
+ if (isBackendAvailable(backend)) {
tableToBeMap.put(cacheKey, backend.getId());
tableToPressureMap.put(tableId,
new
SlidingWindowCounter(table.getGroupCommitIntervalMs() / 1000 + 1));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java
index 28a4decd6be..194bc6a8a59 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java
@@ -223,6 +223,14 @@ public class CloudSystemInfoServiceTest {
String res = infoService.getPhysicalCluster(vcgName);
Assert.assertEquals(pcgName1, res);
+
+ Backend activeBackend = toAdd1.get(1);
+ Assert.assertSame(activeBackend,
+ infoService.getBackendInCurrentCluster(pcgName1,
activeBackend.getId()));
+ Assert.assertSame(activeBackend,
+ infoService.getBackendInCurrentCluster(vcgName,
activeBackend.getId()));
+ Assert.assertNull(infoService.getBackendInCurrentCluster(vcgName,
toAdd2.get(1).getId()));
+ Assert.assertNull(infoService.getBackendInCurrentCluster(vcgName,
Long.MAX_VALUE));
}
// active has 3 dead be and standby has 3 alive be
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerBackendSelectionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerBackendSelectionTest.java
index 7f762f749c9..db26b5ddbfd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerBackendSelectionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerBackendSelectionTest.java
@@ -167,7 +167,8 @@ public class GroupCommitManagerBackendSelectionTest {
backend.setCloudClusterName(cluster);
Mockito.when(cloudSystemInfoService.getCloudIdToBackend(cluster))
.thenReturn(ImmutableMap.of(backend.getId(), backend));
-
Mockito.when(cloudSystemInfoService.getBackend(backend.getId())).thenReturn(backend);
+
Mockito.when(cloudSystemInfoService.getBackendInCurrentCluster(cluster,
backend.getId()))
+ .thenReturn(backend);
BackendSelectionManager.setProviderForTest(policy);
try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerTest.java
new file mode 100644
index 00000000000..7012e16d8de
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerTest.java
@@ -0,0 +1,188 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.system.Backend;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+public class GroupCommitManagerTest {
+ private static final long TABLE_ID = 100L;
+ private static final String VIRTUAL_CLUSTER = "virtual_cluster";
+ private static final String PHYSICAL_CLUSTER_A = "physical_cluster_a";
+ private static final String PHYSICAL_CLUSTER_B = "physical_cluster_b";
+ private static final long BACKEND_A_ID = 10001L;
+ private static final long BACKEND_B_ID = 10002L;
+
+ private String originalCloudUniqueId;
+ private String originalDeployMode;
+ private Env currentEnv;
+ private InternalCatalog internalCatalog;
+ private OlapTable table;
+ private CloudSystemInfoService systemInfoService;
+
+ @Before
+ public void setUp() {
+ originalCloudUniqueId = Config.cloud_unique_id;
+ originalDeployMode = Config.deploy_mode;
+ Config.cloud_unique_id = "test_cloud_unique_id";
+
+ currentEnv = Mockito.mock(Env.class);
+ internalCatalog = Mockito.mock(InternalCatalog.class);
+ table = Mockito.mock(OlapTable.class);
+ systemInfoService = Mockito.mock(CloudSystemInfoService.class);
+
+
Mockito.when(currentEnv.getInternalCatalog()).thenReturn(internalCatalog);
+
Mockito.when(internalCatalog.getTableByTableId(TABLE_ID)).thenReturn(table);
+ Mockito.when(table.getGroupCommitDataBytes()).thenReturn(1024);
+ Mockito.when(table.getGroupCommitIntervalMs()).thenReturn(1000);
+ }
+
+ @After
+ public void tearDown() {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ Config.deploy_mode = originalDeployMode;
+ }
+
+ @Test
+ public void testVirtualComputeGroupUsesActiveBackendsForCacheAndFailover()
throws Exception {
+ Backend backendA = createBackend(BACKEND_A_ID, PHYSICAL_CLUSTER_A);
+ Backend backendB = createBackend(BACKEND_B_ID, PHYSICAL_CLUSTER_B);
+ AtomicReference<ImmutableMap<Long, Backend>> activeBackends =
+ new AtomicReference<>(ImmutableMap.of(BACKEND_A_ID, backendA));
+
+ Mockito.when(systemInfoService.getCloudIdToBackend(VIRTUAL_CLUSTER))
+ .thenAnswer(invocation -> activeBackends.get());
+ Mockito.when(systemInfoService.getBackendInCurrentCluster(
+ Mockito.eq(VIRTUAL_CLUSTER), Mockito.anyLong()))
+ .thenAnswer(invocation ->
activeBackends.get().get(invocation.getArgument(1, Long.class)));
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(currentEnv);
+
mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ GroupCommitManager manager = new GroupCommitManager();
+ Assert.assertEquals(BACKEND_A_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
VIRTUAL_CLUSTER));
+ Mockito.verify(systemInfoService,
Mockito.never()).getPhysicalCluster(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getBackend(Mockito.anyLong());
+
+ Mockito.clearInvocations(systemInfoService);
+ Assert.assertEquals(BACKEND_A_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
VIRTUAL_CLUSTER));
+
Mockito.verify(systemInfoService).getBackendInCurrentCluster(VIRTUAL_CLUSTER,
BACKEND_A_ID);
+ Mockito.verify(systemInfoService,
Mockito.never()).getCloudIdToBackend(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getPhysicalCluster(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getBackend(Mockito.anyLong());
+
+ Mockito.clearInvocations(systemInfoService);
+ activeBackends.set(ImmutableMap.of(BACKEND_B_ID, backendB));
+
+ Assert.assertEquals(BACKEND_B_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
VIRTUAL_CLUSTER));
+
Mockito.verify(systemInfoService).getBackendInCurrentCluster(VIRTUAL_CLUSTER,
BACKEND_A_ID);
+
Mockito.verify(systemInfoService).getCloudIdToBackend(VIRTUAL_CLUSTER);
+ }
+
+ Mockito.verify(systemInfoService,
Mockito.never()).getPhysicalCluster(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getBackend(Mockito.anyLong());
+ }
+
+ @Test
+ public void testLoadDisabledCachedBackendIsReplacedFromActiveBackends()
throws Exception {
+ Backend backendA1 = createBackend(BACKEND_A_ID, PHYSICAL_CLUSTER_A);
+ Backend backendA2 = createBackend(BACKEND_B_ID, PHYSICAL_CLUSTER_A);
+ AtomicReference<ImmutableMap<Long, Backend>> activeBackends =
+ new AtomicReference<>(ImmutableMap.of(BACKEND_A_ID,
backendA1));
+
+ Mockito.when(systemInfoService.getCloudIdToBackend(VIRTUAL_CLUSTER))
+ .thenAnswer(invocation -> activeBackends.get());
+ Mockito.when(systemInfoService.getBackendInCurrentCluster(
+ Mockito.eq(VIRTUAL_CLUSTER), Mockito.anyLong()))
+ .thenAnswer(invocation ->
activeBackends.get().get(invocation.getArgument(1, Long.class)));
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(currentEnv);
+
mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ GroupCommitManager manager = new GroupCommitManager();
+ Assert.assertEquals(BACKEND_A_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
VIRTUAL_CLUSTER));
+
+ Mockito.clearInvocations(systemInfoService);
+ activeBackends.set(ImmutableMap.of(BACKEND_A_ID, backendA1,
BACKEND_B_ID, backendA2));
+ backendA1.setLoadDisabled(true);
+
+ Assert.assertEquals(BACKEND_B_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
VIRTUAL_CLUSTER));
+
Mockito.verify(systemInfoService).getBackendInCurrentCluster(VIRTUAL_CLUSTER,
BACKEND_A_ID);
+
Mockito.verify(systemInfoService).getCloudIdToBackend(VIRTUAL_CLUSTER);
+ }
+
+ Mockito.verify(systemInfoService,
Mockito.never()).getPhysicalCluster(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getBackend(Mockito.anyLong());
+ }
+
+ @Test
+ public void testLocalGroupCommitStillUsesGlobalBackendLookup() throws
Exception {
+ Config.cloud_unique_id = "";
+ Config.deploy_mode = "";
+ Backend backend = createBackend(BACKEND_A_ID, PHYSICAL_CLUSTER_A);
+
+ Mockito.when(systemInfoService.getAllBackendsByAllCluster())
+ .thenReturn(ImmutableMap.of(BACKEND_A_ID, backend));
+
Mockito.when(systemInfoService.getBackend(BACKEND_A_ID)).thenReturn(backend);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(currentEnv);
+
mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService);
+
+ GroupCommitManager manager = new GroupCommitManager();
+ Assert.assertEquals(BACKEND_A_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
null));
+ Assert.assertEquals(BACKEND_A_ID,
+ manager.selectBackendForGroupCommitInternal(TABLE_ID,
null));
+ }
+
+ Mockito.verify(systemInfoService,
Mockito.never()).getPhysicalCluster(Mockito.anyString());
+ Mockito.verify(systemInfoService,
Mockito.never()).getCloudIdToBackend(Mockito.anyString());
+ Mockito.verify(systemInfoService, Mockito.never())
+ .getBackendInCurrentCluster(Mockito.any(), Mockito.anyLong());
+ Mockito.verify(systemInfoService).getBackend(BACKEND_A_ID);
+ }
+
+ private Backend createBackend(long id, String physicalCluster) {
+ Backend backend = new Backend(id, "127.0.0.1", 9050);
+ backend.setCloudClusterName(physicalCluster);
+ backend.setAlive(true);
+ return backend;
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/use_vcg_read_write.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/use_vcg_read_write.groovy
index 0d9404537e1..ae0a37fadc7 100644
---
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/use_vcg_read_write.groovy
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/use_vcg_read_write.groovy
@@ -125,6 +125,10 @@ suite('use_vcg_read_write', 'multi_cluster,docker') {
}
log.info("backends of cluster2: ${clusterName2}
${cluster2Ips}".toString())
+ def groupCommitStreamLoadFe = options.connectToFollower
+ ? cluster.getOneFollowerFe() : cluster.getMasterFe()
+ assertNotNull(groupCommitStreamLoadFe)
+
sql """use @${normalVclusterName}"""
sql """ drop table if exists ${tableName} """
@@ -145,6 +149,9 @@ suite('use_vcg_read_write', 'multi_cluster,docker') {
`k13` datetime NULL
) ENGINE=OLAP
DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ PROPERTIES (
+ "group_commit_interval_ms" = "200"
+ )
"""
sql """
@@ -188,10 +195,12 @@ suite('use_vcg_read_write', 'multi_cluster,docker') {
set 'column_separator', ','
set 'cloud_cluster', 'normalVirtualClusterName'
+ set 'group_commit', 'sync_mode'
+ unset 'label'
file 'all_types.csv'
time 10000 // limit inflight 10s
- setFeAddr cluster.getAllFrontends().get(0).host,
cluster.getAllFrontends().get(0).httpPort
+ setFeAddr groupCommitStreamLoadFe.host,
groupCommitStreamLoadFe.httpPort
check { loadResult, exception, startTime, endTime ->
if (exception != null) {
@@ -370,10 +379,12 @@ suite('use_vcg_read_write', 'multi_cluster,docker') {
set 'column_separator', ','
set 'cloud_cluster', 'normalVirtualClusterName'
+ set 'group_commit', 'sync_mode'
+ unset 'label'
file 'all_types.csv'
time 10000 // limit inflight 10s
- setFeAddr cluster.getAllFrontends().get(0).host,
cluster.getAllFrontends().get(0).httpPort
+ setFeAddr groupCommitStreamLoadFe.host,
groupCommitStreamLoadFe.httpPort
check { loadResult, exception, startTime, endTime ->
if (exception != null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
