From f2016588b93225107e43989dd8468bf32eb893d7 Mon Sep 17 00:00:00 2001 From: Yukang-Lian Date: Fri, 7 Aug 2026 20:18:55 +0800 Subject: [PATCH] [fix](cloud) Fix group commit routing for virtual compute groups ### What problem does this PR solve? Issue Number: None Related PR: #61555 Problem Summary: Group commit resolves a virtual compute group when collecting backend candidates, but previously validated and cached those backends with the original virtual name. Since backends carry the physical compute group name, every healthy candidate was rejected. Resolve the physical compute group once on the master and use it consistently for cache lookup, candidate selection, and membership validation. ### Release note Fix group commit stream loads through virtual compute groups. ### Check List (For Author) - Test - [x] Unit Test - [ ] Regression test (VCG Docker coverage added and loaded locally; the Docker cluster was not run locally) - Behavior changed: - [x] Yes. Group commit requests through a virtual compute group now route to its active physical compute group. - Does this need documentation? - [x] No. --- .../apache/doris/load/GroupCommitManager.java | 14 +- .../doris/load/GroupCommitManagerTest.java | 144 ++++++++++++++++++ .../use_vcg_read_write.groovy | 15 +- 3 files changed, 166 insertions(+), 7 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerTest.java 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 73cb7f2c4f47cb..0aad78bfbbdebc 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 @@ -293,19 +293,22 @@ private long selectBackendForCloudGroupCommitInternal(long tableId, String clust ErrorReport.reportDdlException(ErrorCode.ERR_NO_CLUSTER_ERROR); } - Long cachedBackendId = getCachedBackend(cluster, tableId); + CloudSystemInfoService cloudSystemInfoService = + (CloudSystemInfoService) Env.getCurrentSystemInfo(); + String physicalCluster = cloudSystemInfoService.getPhysicalCluster(cluster); + + Long cachedBackendId = getCachedBackend(physicalCluster, tableId); if (cachedBackendId != null) { return cachedBackendId; } List backends = new ArrayList<>( - ((CloudSystemInfoService) Env.getCurrentSystemInfo()).getCloudIdToBackend(cluster) - .values()); + cloudSystemInfoService.getCloudIdToBackend(physicalCluster).values()); if (backends.isEmpty()) { throw new LoadException("No alive backend"); } // If the cached backend is not active or decommissioned, select a random new backend. - Long randomBackendId = getRandomBackend(cluster, tableId, backends); + Long randomBackendId = getRandomBackend(physicalCluster, tableId, backends); if (randomBackendId != null) { return randomBackendId; } @@ -314,7 +317,8 @@ private long selectBackendForCloudGroupCommitInternal(long tableId, String clust + ", decommissioned=" + be.isDecommissioned() + ", decommissioning=" + be.isDecommissioning() + " }") .collect(Collectors.toList()); - throw new LoadException("No suitable backend for cloud cluster=" + cluster + ", backends = " + backendsInfo); + throw new LoadException("No suitable backend for cloud cluster=" + cluster + + ", physical cluster=" + physicalCluster + ", backends = " + backendsInfo); } private long selectBackendForLocalGroupCommitInternal(long tableId) throws LoadException { 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 00000000000000..6aac22faa5490b --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/load/GroupCommitManagerTest.java @@ -0,0 +1,144 @@ +// 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 Env currentEnv; + private InternalCatalog internalCatalog; + private OlapTable table; + private CloudSystemInfoService systemInfoService; + + @Before + public void setUp() { + originalCloudUniqueId = Config.cloud_unique_id; + 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; + } + + @Test + public void testVirtualComputeGroupUsesPhysicalClusterForCacheAndFailover() throws Exception { + Backend backendA = createBackend(BACKEND_A_ID, PHYSICAL_CLUSTER_A); + Backend backendB = createBackend(BACKEND_B_ID, PHYSICAL_CLUSTER_B); + AtomicReference activePhysicalCluster = new AtomicReference<>(PHYSICAL_CLUSTER_A); + + Mockito.when(systemInfoService.getPhysicalCluster(VIRTUAL_CLUSTER)) + .thenAnswer(invocation -> activePhysicalCluster.get()); + Mockito.when(systemInfoService.getCloudIdToBackend(PHYSICAL_CLUSTER_A)) + .thenReturn(ImmutableMap.of(BACKEND_A_ID, backendA)); + Mockito.when(systemInfoService.getCloudIdToBackend(PHYSICAL_CLUSTER_B)) + .thenReturn(ImmutableMap.of(BACKEND_B_ID, backendB)); + Mockito.when(systemInfoService.getBackend(BACKEND_A_ID)).thenReturn(backendA); + Mockito.when(systemInfoService.getBackend(BACKEND_B_ID)).thenReturn(backendB); + + try (MockedStatic 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)); + Assert.assertEquals(BACKEND_A_ID, + manager.selectBackendForGroupCommitInternal(TABLE_ID, VIRTUAL_CLUSTER)); + + activePhysicalCluster.set(PHYSICAL_CLUSTER_B); + + Assert.assertEquals(BACKEND_B_ID, + manager.selectBackendForGroupCommitInternal(TABLE_ID, VIRTUAL_CLUSTER)); + } + + Mockito.verify(systemInfoService, Mockito.times(3)).getPhysicalCluster(VIRTUAL_CLUSTER); + Mockito.verify(systemInfoService).getCloudIdToBackend(PHYSICAL_CLUSTER_A); + Mockito.verify(systemInfoService).getCloudIdToBackend(PHYSICAL_CLUSTER_B); + Mockito.verify(systemInfoService, Mockito.never()).getCloudIdToBackend(VIRTUAL_CLUSTER); + Mockito.verify(systemInfoService).getBackend(BACKEND_A_ID); + } + + @Test + public void testLoadDisabledCachedBackendIsReplacedInPhysicalCluster() throws Exception { + Backend backendA1 = createBackend(BACKEND_A_ID, PHYSICAL_CLUSTER_A); + Backend backendA2 = createBackend(BACKEND_B_ID, PHYSICAL_CLUSTER_A); + + Mockito.when(systemInfoService.getPhysicalCluster(VIRTUAL_CLUSTER)).thenReturn(PHYSICAL_CLUSTER_A); + Mockito.when(systemInfoService.getCloudIdToBackend(PHYSICAL_CLUSTER_A)) + .thenReturn(ImmutableMap.of(BACKEND_A_ID, backendA1)) + .thenReturn(ImmutableMap.of(BACKEND_A_ID, backendA1, BACKEND_B_ID, backendA2)); + Mockito.when(systemInfoService.getBackend(BACKEND_A_ID)).thenReturn(backendA1); + + try (MockedStatic 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)); + + backendA1.setLoadDisabled(true); + + Assert.assertEquals(BACKEND_B_ID, + manager.selectBackendForGroupCommitInternal(TABLE_ID, VIRTUAL_CLUSTER)); + } + + Mockito.verify(systemInfoService, Mockito.times(2)).getCloudIdToBackend(PHYSICAL_CLUSTER_A); + } + + 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 0d9404537e17b5..ae0a37fadc7513 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) {