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