[fix](cloud) Fix group commit routing for virtual compute groups

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<Backend> 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 {

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<String> 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<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));
+            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<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));
+
+            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;
+    }
+}

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) {