This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 14d3d957c02 branch-4.1: [fix](group commit) Fix group commit routing
for virtual compute groups #66585 (#67590)
14d3d957c02 is described below
commit 14d3d957c02f79e5505879f9de60b3a4a2a05f1c
Author: Jamie <[email protected]>
AuthorDate: Thu Sep 10 11:37:04 2026 +0800
branch-4.1: [fix](group commit) Fix group commit routing for virtual
compute groups #66585 (#67590)
### What problem does this PR solve?
Issue Number: None
Related PR: #66585
Problem Summary: Backport #66585 to `branch-4.1`.
Group Commit asks `CloudSystemInfoService` for the backend pool selected
by a physical or virtual compute group, but then compared those physical
backends with the original virtual name. That rejected every healthy
candidate and could also reuse a cached backend after it left the
current pool.
This backport keeps virtual compute group routing inside
`CloudSystemInfoService`: cache hits use a scoped lookup in the current
selected pool, while random selection trusts the already scoped
candidate pool and performs only health checks. The implementation is
adapted to the branch-4.1 Group Commit and cloud-system APIs while
preserving the branch's existing pressure counter.
### Release note
Fix Group Commit loads routed through virtual compute groups.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- Focused tests are included; hosted CI is requested with `run
buildall`.
- `build-support/check-format.sh`: passed
- `git diff --check`: passed
- [ ] Manual test
- [ ] No need to test or manual test
- Behavior changed: Yes. Group Commit now accepts healthy backends
returned for a virtual compute group and invalidates cached backends
outside its current selected pool.
- Does this need documentation? No
---
.../doris/cloud/system/CloudSystemInfoService.java | 41 +++++
.../org/apache/doris/load/GroupCommitManager.java | 60 +++++--
.../cloud/system/CloudSystemInfoServiceTest.java | 8 +
.../GroupCommitManagerBackendSelectionTest.java | 3 +-
.../apache/doris/load/GroupCommitManagerTest.java | 188 +++++++++++++++++++++
.../use_vcg_read_write.groovy | 15 +-
6 files changed, 297 insertions(+), 18 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 1e2cd8cfa41..104489572aa 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
@@ -1406,6 +1406,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 46c64e7f3bc..509c7a7cd0d 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
@@ -320,7 +320,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;
}
@@ -390,8 +390,17 @@ public class GroupCommitManager {
// Maybe one thread removes the tableId from the tableToBeMap.
// Another thread gets the same tableId but can not find this
tableId.
// So another thread needs to get the random backend.
- 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)) {
+ debugBlockAfterCachedBackendAvailable(cacheKey, tableId,
backendId);
return backend.getId();
} else {
tableToBeMap.invalidate(cacheKey);
@@ -403,23 +412,44 @@ 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;
+ private static void debugBlockAfterCachedBackendAvailable(
+ String cacheKey, long tableId, long backendId) {
+ final String debugPointName =
"GroupCommitManager.getCachedBackend.afterAvailable.block";
+ DebugPointUtil.DebugPoint debugPoint =
DebugPointUtil.getDebugPoint(debugPointName);
+ if (debugPoint == null) {
+ return;
}
- if (!Config.isCloudMode()) {
- return true;
+ long expectedTableId = debugPoint.param("table_id", -1L);
+ long expectedBackendId = debugPoint.param("backend_id", -1L);
+ if ((expectedTableId != -1L && expectedTableId != tableId)
+ || (expectedBackendId != -1L && expectedBackendId !=
backendId)) {
+ return;
}
- return cloudCluster == null ||
cloudCluster.equals(backend.getCloudClusterName());
+ LOG.info("debug block after cached group commit backend passed
availability check, "
+ + "cacheKey={}, tableId={}, backendId={}", cacheKey,
tableId, backendId);
+ while (DebugPointUtil.isEnable(debugPointName)) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("group commit cached backend debug block interrupted,
tableId={}, backendId={}",
+ tableId, backendId);
+ return;
+ }
+ }
+ }
+
+ 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
@@ -436,14 +466,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]
