You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/10/15 02:18:37 UTC
(doris) 12/12: branch-4.1: [fix](cloud) Preserve recreated compute group mapping #67913 (#68939)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 7b81582c97686c7c4df8fa63dda576123dbe8b39
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Thu Oct 15 09:32:52 2026 +0800
branch-4.1: [fix](cloud) Preserve recreated compute group mapping #67913 (#68939)
Cherry-picked from #67913
---------
Co-authored-by: deardeng <de...@selectdb.com>
---
.../doris/cloud/catalog/CloudClusterChecker.java | 25 +-
.../cloud/catalog/CloudInstanceStatusChecker.java | 59 +++-
.../apache/doris/cloud/catalog/CloudReplica.java | 2 +-
.../doris/cloud/system/CloudSystemInfoService.java | 74 +++--
.../cloud/catalog/CloudClusterCheckerTest.java | 308 +++++++++++++++++++++
.../cloud/system/CloudSystemInfoServiceTest.java | 44 +++
.../test_compute_group_name_reconciliation.out | 21 ++
.../test_compute_group_name_reconciliation.groovy | 169 +++++++++++
8 files changed, 651 insertions(+), 51 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java
index c8eee36f505..c38bc7debc8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java
@@ -27,6 +27,7 @@ import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.metric.MetricRepo;
@@ -146,19 +147,22 @@ public class CloudClusterChecker extends MasterDaemon {
if (LOG.isDebugEnabled()) {
LOG.debug("begin to drop clusterId: {}", delId);
}
- String delClusterName = cloudSystemInfoService.getClusterNameByClusterId(delId);
- if (delClusterName.isEmpty()) {
- return;
- }
- ((CloudEnv) Env.getCurrentEnv()).getCacheHotspotMgr().cancelTableFilterJobsForClusterChange(
- delClusterName, "system cancel: compute group " + delClusterName + " dropped");
List<Backend> toDel =
new ArrayList<>(finalClusterIdToBackend.getOrDefault(delId, new ArrayList<>()));
+ // The name index may already belong to a same-name replacement. Use the
+ // obsolete group's own BE tags, and never skip ID cleanup for a missing name.
+ String delClusterName = toDel.stream().map(Backend::getCloudClusterName).findFirst()
+ .orElseGet(() -> cloudSystemInfoService.getClusterNameByClusterId(delId));
+ // Name-scoped jobs may already belong to the replacement group.
+ if (delId.equals(cloudSystemInfoService.getCloudClusterIdByName(delClusterName))) {
+ ((CloudEnv) Env.getCurrentEnv()).getCacheHotspotMgr().cancelTableFilterJobsForClusterChange(
+ delClusterName, "system cancel: compute group " + delClusterName + " dropped");
+ }
cloudSystemInfoService.updateCloudBackends(new ArrayList<>(), toDel);
// del clusterName
// del clusterID
MetricRepo.unregisterCloudMetrics(delId, delClusterName, toDel);
- cloudSystemInfoService.dropCluster(delId, delClusterName);
+ cloudSystemInfoService.removeComputeGroup(delId, delClusterName);
}
);
}
@@ -539,6 +543,11 @@ public class CloudClusterChecker extends MasterDaemon {
}
private void checkCloudBackends() {
+ if (DebugPointUtil.isEnable("CloudClusterChecker.checkCloudBackends.pause")) {
+ LOG.info("CloudClusterChecker.checkCloudBackends.pause phase={}", DebugPointUtil.getDebugParamOrDefault(
+ "CloudClusterChecker.checkCloudBackends.pause", "phase", ""));
+ return;
+ }
Map<String, List<Backend>> clusterIdToBackend = cloudSystemInfoService.getCloudClusterIdToBackend(false);
//rpc to ms, to get mysql user can use cluster_id
// NOTE: rpc args all empty, use cluster_unique_id to get a instance's all cluster info.
@@ -574,6 +583,8 @@ public class CloudClusterChecker extends MasterDaemon {
// clusterID local == remote, diff nodes
checkDiffNode(remoteClusterIdToPB, clusterIdToBackend);
+ cloudSystemInfoService.refreshComputeGroupNames(remoteClusterIdToPB.values());
+
// check mem map
checkFeNodesMapValid();
} catch (Exception e) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
index cc9d256d566..72742911daf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
@@ -26,6 +26,7 @@ import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.nereids.trees.plans.commands.WarmUpClusterCommand;
@@ -58,6 +59,12 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
@Override
protected void runAfterCatalogReady() {
try {
+ if (DebugPointUtil.isEnable("CloudInstanceStatusChecker.runAfterCatalogReady.pause")) {
+ LOG.info("CloudInstanceStatusChecker.runAfterCatalogReady.pause phase={}",
+ DebugPointUtil.getDebugParamOrDefault(
+ "CloudInstanceStatusChecker.runAfterCatalogReady.pause", "phase", ""));
+ return;
+ }
long start = System.currentTimeMillis();
Cloud.GetInstanceResponse response = cloudSystemInfoService.getCloudInstance();
if (!isResponseValid(response)) {
@@ -65,6 +72,17 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
}
Cloud.InstanceInfoPB instance = response.getInstance();
+ if (DebugPointUtil.isEnable("CloudInstanceStatusChecker.afterGetInstance.pause")
+ && instance.getClustersList().stream().anyMatch(c -> c.getClusterId().equals(
+ DebugPointUtil.getDebugParamOrDefault(
+ "CloudInstanceStatusChecker.afterGetInstance.pause", "cluster_id", "")))) {
+ LOG.info("CloudInstanceStatusChecker.afterGetInstance.pause phase={}",
+ DebugPointUtil.getDebugParamOrDefault(
+ "CloudInstanceStatusChecker.afterGetInstance.pause", "phase", ""));
+ while (DebugPointUtil.isEnable("CloudInstanceStatusChecker.afterGetInstance.pause")) {
+ Thread.sleep(100);
+ }
+ }
cloudSystemInfoService.setInstanceStatus(instance.getStatus());
syncStorageVault(instance);
processVirtualClusters(instance.getClustersList());
@@ -107,8 +125,10 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
List<Cloud.ClusterPB> computeClusters = new ArrayList<>();
categorizeClusters(clusters, virtualClusters, computeClusters);
handleComputeClusters(computeClusters);
- handleVirtualClusters(virtualClusters, computeClusters);
+ List<Cloud.ClusterPB> reconciledVirtualClusters = handleVirtualClusters(virtualClusters, computeClusters);
+ // A rejected update still proves that the group exists in MS, so use the full snapshot for removal.
removeObsoleteVirtualGroups(virtualClusters);
+ cloudSystemInfoService.refreshComputeGroupNames(reconciledVirtualClusters);
}
private void handleComputeClusters(List<Cloud.ClusterPB> computeClusters) {
@@ -179,15 +199,23 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
}
}
- private void handleVirtualClusters(List<Cloud.ClusterPB> virtualGroups, List<Cloud.ClusterPB> computeClusters) {
+ private List<Cloud.ClusterPB> handleVirtualClusters(
+ List<Cloud.ClusterPB> virtualGroups, List<Cloud.ClusterPB> computeClusters) {
+ List<Cloud.ClusterPB> reconciledGroups = new ArrayList<>();
for (Cloud.ClusterPB virtualGroupInMs : virtualGroups) {
CloudComputeGroupMeta virtualGroupInFe = cloudSystemInfoService
.getComputeGroupById(virtualGroupInMs.getClusterId());
+ boolean reconciled;
if (virtualGroupInFe != null) {
- handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe);
+ reconciled = handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe);
} else {
- handleNewVirtualComputeGroup(virtualGroupInMs, computeClusters);
+ reconciled = handleNewVirtualComputeGroup(virtualGroupInMs, computeClusters);
+ }
+ // Rejected renames must not publish a name for metadata whose update was rejected.
+ if (!reconciled) {
+ continue;
}
+ reconciledGroups.add(virtualGroupInMs);
// just fe master gen file cache sync task
if (Env.getCurrentEnv().isMaster()) {
// get again in fe mem
@@ -201,6 +229,7 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
syncFileCacheTasksForVirtualGroup(virtualGroupInMs, virtualGroupInFe);
}
}
+ return reconciledGroups;
}
private void cancelCacheJobs(CloudComputeGroupMeta vcgInFe, List<String> jobIds) {
@@ -349,21 +378,22 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
}
}
- private void handleExistingVirtualComputeGroup(
+ private boolean handleExistingVirtualComputeGroup(
Cloud.ClusterPB clusterInMs, CloudComputeGroupMeta virtualGroupInFe) {
if (!isClusterIdConsistent(clusterInMs, virtualGroupInFe)) {
- return;
+ return false;
}
if (!isClusterPolicyValid(clusterInMs)) {
- return;
+ return false;
}
if (!areSubComputeGroupsValid(clusterInMs, virtualGroupInFe)) {
- return;
+ return false;
}
diffAndUpdateComputeGroup(clusterInMs, virtualGroupInFe);
+ return true;
}
private boolean isClusterIdConsistent(Cloud.ClusterPB cluster, CloudComputeGroupMeta computeGroup) {
@@ -478,27 +508,27 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
}
}
- private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List<Cloud.ClusterPB> computeClusters) {
+ private boolean handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List<Cloud.ClusterPB> computeClusters) {
List<String> subComputeGroups = cluster.getClusterNamesList();
if (subComputeGroups.isEmpty()) {
LOG.info("found virtual cluster {} which has no sub clusters, skip empty virtual cluster", cluster);
- return;
+ return false;
}
if (subComputeGroups.size() != 2) {
LOG.warn("virtual compute err, sub compute group size not eq 2, in ms {}", subComputeGroups);
- return;
+ return false;
}
if (!cluster.hasClusterPolicy()) {
LOG.warn("virtual compute err, no cluster policy {}", cluster);
- return;
+ return false;
}
if (!cluster.getClusterPolicy().hasActiveClusterName()) {
LOG.warn("virtual compute err, active cluster empty in ms {}", cluster);
- return;
+ return false;
}
if (cluster.getClusterPolicy().getStandbyClusterNamesList().size() != 1) {
LOG.warn("virtual compute err, standby cluster size not eq 1 in ms {}", cluster);
- return;
+ return false;
}
checkSubClusters(subComputeGroups, cluster, computeClusters);
CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta(cluster.getClusterId(),
@@ -513,6 +543,7 @@ public class CloudInstanceStatusChecker extends MasterDaemon {
computeGroup.setNeedRebuildFileCache(true);
cloudSystemInfoService.addComputeGroup(cluster.getClusterId(), computeGroup);
MetricRepo.registerCloudMetrics(cluster.getClusterId(), cluster.getClusterName());
+ return true;
}
private void checkSubClusters(List<String> subClusterNames, Cloud.ClusterPB cluster,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
index 4c51c65457e..4c4bcb175a8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
@@ -643,7 +643,7 @@ public class CloudReplica extends Replica implements GsonPostProcessable {
*
* Such an entry is already dead weight: getBackendIdImpl() resolves the backend id, gets null and
* falls back to hashReplicaToBe(), so removing it does not change routing. But nothing ever removes
- * it either -- dropCluster() only touches CloudSystemInfoService, and the rebalancer only walks the
+ * it either -- removeComputeGroup() only touches CloudSystemInfoService, and the rebalancer only walks the
* compute groups that currently exist -- so entries of dropped compute groups pile up forever, both
* in FE heap and in the image (the `bes`/`be` field).
*
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 104489572aa..05c4e04e7de 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
@@ -68,6 +68,7 @@ import org.apache.logging.log4j.Logger;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
@@ -249,17 +250,46 @@ public class CloudSystemInfoService extends SystemInfoService {
clusterIdToBackend.computeIfAbsent(clusterId, k -> new ArrayList<>());
}
- public void removeVirtualClusterInfoFromMapsNoLock(String clusterId, String clusterName) {
- LOG.info("remove virtual cluster info from maps, clusterId={}, clusterName={}", clusterId, clusterName);
- clusterIdToBackend.remove(clusterId);
- clusterNameToId.remove(clusterName);
- }
-
public void renameVirtualClusterInfoFromMapsNoLock(String clusterId, String oldClusterName, String newClusterName) {
LOG.info("remove virtual cluster info from maps, clusterId={}, name from {} to {}",
clusterId, oldClusterName, newClusterName);
clusterNameToId.put(newClusterName, clusterId);
- clusterNameToId.remove(oldClusterName);
+ clusterNameToId.remove(oldClusterName, clusterId);
+ }
+
+ // Physical and virtual group checkers fetch and apply snapshots independently.
+ // For example:
+ // 1. A current physical snapshot installs name -> newId.
+ // 2. A delayed virtual snapshot overwrites it with name -> oldId.
+ // 3. A later virtual snapshot removes oldId and its name mapping.
+ // The new physical group's metadata and BEs still exist, but its name mapping
+ // is missing. Normal add/rename detection cannot repair it because neither
+ // the group ID nor the BE names have changed.
+ //
+ // Reconcile name mappings for locally installed groups on every sync cycle.
+ // Callers must exclude records whose metadata reconciliation was rejected.
+ // This does not reject stale snapshots; it restores the current mapping once
+ // a later cycle applies current metadata from the meta service.
+ public void refreshComputeGroupNames(Collection<Cloud.ClusterPB> remoteComputeGroups) {
+ wlock.lock();
+ try {
+ for (Cloud.ClusterPB group : remoteComputeGroups) {
+ String id = group.getClusterId();
+ // Empty physical groups and rejected virtual groups may not be installed locally.
+ // Do not create a name pointing to missing metadata.
+ if (!computeGroupIdToComputeGroup.containsKey(id)) {
+ continue;
+ }
+ String name = group.getClusterName();
+ String previousId = clusterNameToId.put(name, id);
+ if (!id.equals(previousId)) {
+ LOG.warn("repair compute group name mapping from meta service, name={}, oldId={}, currentId={}",
+ name, previousId, id);
+ }
+ }
+ } finally {
+ wlock.unlock();
+ }
}
public CloudComputeGroupMeta getComputeGroupByName(String computeGroupName) {
@@ -423,11 +453,17 @@ public class CloudSystemInfoService extends SystemInfoService {
}
}
+ // Remove local metadata for a physical or virtual compute group. Physical backends
+ // must be removed by the caller before removing the group.
public void removeComputeGroup(String computeGroupId, String computeGroupName) {
try {
wlock.lock();
+ LOG.info("remove compute group, id={}, name={}", computeGroupId, computeGroupName);
computeGroupIdToComputeGroup.remove(computeGroupId);
- removeVirtualClusterInfoFromMapsNoLock(computeGroupId, computeGroupName);
+ clusterIdToBackend.remove(computeGroupId);
+ // Earlier checkers could publish aliases from rejected renames. Remove every name
+ // still owned by this ID, without removing names already reused by another group.
+ clusterNameToId.entrySet().removeIf(entry -> computeGroupId.equals(entry.getValue()));
invalidateCloudColocatePlacement(computeGroupId);
} finally {
wlock.unlock();
@@ -634,22 +670,13 @@ public class CloudSystemInfoService extends SystemInfoService {
be = be.stream().filter(i -> !d.contains(i.getId())).collect(Collectors.toList());
// ATTN: clusterId may have zero nodes
clusterIdToBackend.replace(clusterId, be);
- // such as dropCluster, but no lock
// ATTN: Empty clusters are treated as dropped clusters.
if (be.isEmpty()) {
LOG.info("del clusterId {} and clusterName {} due to be nodes eq 0", clusterId, clusterName);
MetricRepo.unregisterCloudMetrics(clusterId, clusterName, toDel);
- boolean succ = clusterNameToId.remove(clusterName, clusterId);
// remove from computeGroupIdToComputeGroup
removeComputeGroup(clusterId, clusterName);
-
- if (!succ) {
- LOG.warn("impossible, somewhere err, clusterNameToId {}, "
- + "want remove cluster name {}, cluster id {}",
- clusterNameToId, clusterName, clusterId);
- }
- clusterIdToBackend.remove(clusterId);
}
LOG.info("update (del) cloud cluster map, clusterName={} clusterId={} backendNum={} current backend={}",
clusterName, clusterId, be.size(), b);
@@ -1232,7 +1259,7 @@ public class CloudSystemInfoService extends SystemInfoService {
final String originalName, final String clusterId) {
wlock.lock();
try {
- clusterNameToId.remove(originalName);
+ clusterNameToId.remove(originalName, clusterId);
clusterNameToId.put(newName, clusterId);
} finally {
wlock.unlock();
@@ -1264,17 +1291,6 @@ public class CloudSystemInfoService extends SystemInfoService {
return clusterName;
}
- public void dropCluster(final String clusterId, final String clusterName) {
- wlock.lock();
- try {
- clusterNameToId.remove(clusterName, clusterId);
- clusterIdToBackend.remove(clusterId);
- invalidateCloudColocatePlacement(clusterId);
- } finally {
- wlock.unlock();
- }
- }
-
public List<String> getCloudClusterNames() {
rlock.lock();
try {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java
new file mode 100644
index 00000000000..2e71cfcfb98
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java
@@ -0,0 +1,308 @@
+// 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.cloud.catalog;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.CacheHotspotManager;
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.persist.EditLog;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+class CloudClusterCheckerTest {
+ private String originalCloudUniqueId;
+ private CloudSystemInfoService service;
+ private CloudEnv env;
+ private MockedStatic<Env> mockedEnv;
+
+ @BeforeEach
+ void setUp() {
+ originalCloudUniqueId = Config.cloud_unique_id;
+ Config.cloud_unique_id = "test_cloud";
+ service = Mockito.spy(new CloudSystemInfoService());
+ env = Mockito.mock(CloudEnv.class);
+ AtomicLong nextId = new AtomicLong(10000);
+ Mockito.when(env.getNextId()).thenAnswer(invocation -> nextId.incrementAndGet());
+ Mockito.when(env.getEditLog()).thenReturn(Mockito.mock(EditLog.class));
+ Mockito.when(env.getCacheHotspotMgr()).thenReturn(Mockito.mock(CacheHotspotManager.class));
+ mockedEnv = mockEnv();
+ }
+
+ private MockedStatic<Env> mockEnv() {
+ MockedStatic<Env> mock = Mockito.mockStatic(Env.class);
+ mock.when(Env::getCurrentEnv).thenReturn(env);
+ mock.when(Env::getCurrentSystemInfo).thenReturn(service);
+ return mock;
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockedEnv.close();
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+
+ @Test
+ void testCheckerDeletesRecreatedGroupById() throws Exception {
+ Cloud.ClusterPB oldGroup = physical("old", "reused", "127.0.0.1");
+ Cloud.ClusterPB newGroup = physical("new", "reused", "127.0.0.2");
+ syncPhysical(oldGroup);
+ long oldBackendId = service.getCloudClusterIdToBackend(false).get("old").get(0).getId();
+
+ syncPhysical(newGroup);
+
+ Assertions.assertNull(service.getBackend(oldBackendId));
+ Assertions.assertNull(service.getComputeGroupById("old"));
+ Assertions.assertFalse(service.getCloudClusterIdToBackend(false).containsKey("old"));
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ syncPhysical(newGroup);
+ Assertions.assertEquals(1, service.getAllBackendsByAllCluster().size());
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ }
+
+ @Test
+ void testCheckerRenamesGroupAfterOldNameIsReused() throws Exception {
+ Cloud.ClusterPB oldGroup = physical("old", "reused", "127.0.0.1");
+ Cloud.ClusterPB renamed = oldGroup.toBuilder().setClusterName("renamed").build();
+ Cloud.ClusterPB replacement = physical("new", "reused", "127.0.0.2");
+ syncPhysical(oldGroup);
+ syncPhysical(renamed, replacement);
+
+ Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed"));
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ syncPhysical(renamed, replacement);
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ }
+
+ @Test
+ void testPhysicalRenameDoesNotRemoveAnotherGroupsName() {
+ service.addComputeGroup("old", new CloudComputeGroupMeta("old", "reused",
+ CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE));
+ service.addComputeGroup("new", new CloudComputeGroupMeta("new", "reused",
+ CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE));
+ service.updateClusterNameToId("renamed", "reused", "old");
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed"));
+ }
+
+ @Test
+ void testVirtualRenameDoesNotRemoveAnotherGroupsName() {
+ service.addComputeGroup("old", new CloudComputeGroupMeta("old", "reused",
+ CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL));
+ service.addComputeGroup("new", new CloudComputeGroupMeta("new", "reused",
+ CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL));
+ service.renameVirtualComputeGroup("old", "reused", new CloudComputeGroupMeta("old", "renamed",
+ CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL));
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed"));
+ }
+
+ @Test
+ void testLaterCycleRepairsNameAfterStaleVirtualSnapshot() throws Exception {
+ Cloud.ClusterPB active = physical("active", "active", "127.0.0.1");
+ Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2");
+ Cloud.ClusterPB replacement = physical("new", "reused", "127.0.0.3");
+ syncPhysical(active, standby);
+ Cloud.GetInstanceResponse stale = instance(active, standby, virtual("old", "reused"));
+ CountDownLatch responseFetched = new CountDownLatch(1);
+ CountDownLatch applyResponse = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ responseFetched.countDown();
+ Assertions.assertTrue(applyResponse.await(30, TimeUnit.SECONDS));
+ return stale;
+ }).when(service).getCloudInstance();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> staleCycle = executor.submit(() -> {
+ try (MockedStatic<Env> ignored = mockEnv()) {
+ new CloudInstanceStatusChecker(service).runAfterCatalogReady();
+ }
+ });
+ Assertions.assertTrue(responseFetched.await(30, TimeUnit.SECONDS));
+ syncPhysical(active, standby, replacement);
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ applyResponse.countDown();
+ staleCycle.get(30, TimeUnit.SECONDS);
+ Assertions.assertEquals("old", service.getCloudClusterIdByName("reused"));
+
+ // A current physical snapshot must repair even a non-empty, wrong mapping.
+ syncPhysical(active, standby, replacement);
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ syncInstance(active, standby, replacement);
+ Assertions.assertNull(service.getComputeGroupById("old"));
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ syncPhysical(active, standby, replacement);
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+ } finally {
+ applyResponse.countDown();
+ executor.shutdownNow();
+ Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void testLaterCyclesRepairMissingPhysicalAndVirtualNames() throws Exception {
+ Cloud.ClusterPB active = physical("active", "active", "127.0.0.1");
+ Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2");
+ Cloud.ClusterPB current = physical("new", "reused", "127.0.0.3");
+ syncPhysical(active, standby, current);
+ syncInstance(active, standby, virtual("old", "reused"));
+ syncInstance(active, standby, current);
+ Assertions.assertNull(service.getCloudClusterIdByName("reused"));
+ syncPhysical(active, standby, current);
+ Assertions.assertEquals("new", service.getCloudClusterIdByName("reused"));
+
+ // Reverse the types: an obsolete physical snapshot overwrites a current virtual group.
+ syncPhysical(active, standby);
+ Cloud.ClusterPB currentVirtual = virtual("virtual", "reused");
+ syncInstance(active, standby, currentVirtual);
+ syncPhysical(active, standby, current);
+ syncPhysical(active, standby);
+ Assertions.assertNull(service.getCloudClusterIdByName("reused"));
+ syncInstance(active, standby, currentVirtual);
+ Assertions.assertEquals("virtual", service.getCloudClusterIdByName("reused"));
+ syncInstance(active, standby, currentVirtual);
+ Assertions.assertEquals("virtual", service.getCloudClusterIdByName("reused"));
+ Assertions.assertNull(service.getComputeGroupById("new"));
+ }
+
+ @Test
+ void testRejectedVirtualPolicyDoesNotPublishRenamedGroup() throws Exception {
+ assertRejectedVirtualRename(virtual("virtual", "renamed").toBuilder()
+ .clearClusterPolicy().build());
+ }
+
+ @Test
+ void testRejectedVirtualSubgroupsDoNotPublishRenamedGroup() throws Exception {
+ assertRejectedVirtualRename(virtual("virtual", "renamed").toBuilder()
+ .clearClusterNames().addClusterNames("active").build());
+ }
+
+ private void assertRejectedVirtualRename(Cloud.ClusterPB rejected) throws Exception {
+ Cloud.ClusterPB active = physical("active", "active", "127.0.0.1");
+ Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2");
+ syncPhysical(active, standby);
+ syncInstance(active, standby, virtual("virtual", "original"));
+
+ syncInstance(active, standby, rejected);
+
+ Assertions.assertEquals("original", service.getComputeGroupById("virtual").getName());
+ Assertions.assertEquals("virtual", service.getCloudClusterIdByName("original"));
+ Assertions.assertNull(service.getCloudClusterIdByName("renamed"));
+
+ // The rejected record must not make the still-present group look obsolete.
+ // Once it actually disappears from MS, neither name may point to its removed ID.
+ syncInstance(active, standby);
+ Assertions.assertNull(service.getComputeGroupById("virtual"));
+ Assertions.assertNull(service.getCloudClusterIdByName("original"));
+ Assertions.assertNull(service.getCloudClusterIdByName("renamed"));
+ }
+
+ @Test
+ void testAcceptedVirtualRenamePublishesNewName() throws Exception {
+ Cloud.ClusterPB active = physical("active", "active", "127.0.0.1");
+ Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2");
+ syncPhysical(active, standby);
+ syncInstance(active, standby, virtual("virtual", "original"));
+
+ syncInstance(active, standby, virtual("virtual", "renamed"));
+
+ Assertions.assertEquals("renamed", service.getComputeGroupById("virtual").getName());
+ Assertions.assertEquals("virtual", service.getCloudClusterIdByName("renamed"));
+ Assertions.assertNull(service.getCloudClusterIdByName("original"));
+ }
+
+ @Test
+ void testRejectedNewVirtualGroupDoesNotPublishName() {
+ syncInstance(virtual("virtual", "rejected").toBuilder().clearClusterPolicy().build());
+
+ Assertions.assertNull(service.getComputeGroupById("virtual"));
+ Assertions.assertNull(service.getCloudClusterIdByName("rejected"));
+ }
+
+ @Test
+ void testVirtualRemovalClearsStaleAliasesButKeepsReusedName() throws Exception {
+ Cloud.ClusterPB active = physical("active", "active", "127.0.0.1");
+ Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2");
+ syncPhysical(active, standby);
+ syncInstance(active, standby, virtual("virtual", "reused"));
+ // Simulate an alias left by an earlier checker that refreshed a rejected rename.
+ service.addVirtualClusterInfoToMapsNoLock("virtual", "stale_alias");
+ syncPhysical(active, standby, physical("replacement", "reused", "127.0.0.3"));
+
+ syncInstance(active, standby);
+
+ Assertions.assertNull(service.getComputeGroupById("virtual"));
+ Assertions.assertNull(service.getCloudClusterIdByName("stale_alias"));
+ Assertions.assertEquals("replacement", service.getCloudClusterIdByName("reused"));
+ }
+
+ private void syncPhysical(Cloud.ClusterPB... groups) throws Exception {
+ Mockito.doReturn(Cloud.GetClusterResponse.newBuilder().setStatus(ok())
+ .addAllCluster(Arrays.asList(groups)).build()).when(service).getCloudCluster("", "", "");
+ // The production reconciliation entry, including add, delete, diff and validation.
+ Method method = CloudClusterChecker.class.getDeclaredMethod("checkCloudBackends");
+ method.setAccessible(true);
+ method.invoke(new CloudClusterChecker(service));
+ }
+
+ private void syncInstance(Cloud.ClusterPB... groups) {
+ Mockito.doReturn(instance(groups)).when(service).getCloudInstance();
+ new CloudInstanceStatusChecker(service).runAfterCatalogReady();
+ }
+
+ private Cloud.GetInstanceResponse instance(Cloud.ClusterPB... groups) {
+ return Cloud.GetInstanceResponse.newBuilder().setStatus(ok())
+ .setInstance(Cloud.InstanceInfoPB.newBuilder().setStatus(Cloud.InstanceInfoPB.Status.NORMAL)
+ .addAllClusters(Arrays.asList(groups))).build();
+ }
+
+ private Cloud.MetaServiceResponseStatus ok() {
+ return Cloud.MetaServiceResponseStatus.newBuilder().setCode(Cloud.MetaServiceCode.OK).build();
+ }
+
+ private Cloud.ClusterPB physical(String id, String name, String ip) {
+ return Cloud.ClusterPB.newBuilder().setClusterId(id).setClusterName(name)
+ .setType(Cloud.ClusterPB.Type.COMPUTE).setClusterStatus(Cloud.ClusterStatus.NORMAL)
+ .addNodes(Cloud.NodeInfoPB.newBuilder().setIp(ip).setHeartbeatPort(9050)
+ .setCloudUniqueId(id).setStatus(Cloud.NodeStatusPB.NODE_STATUS_RUNNING)).build();
+ }
+
+ private Cloud.ClusterPB virtual(String id, String name) {
+ return Cloud.ClusterPB.newBuilder().setClusterId(id).setClusterName(name)
+ .setType(Cloud.ClusterPB.Type.VIRTUAL).addAllClusterNames(Arrays.asList("active", "standby"))
+ .setClusterPolicy(Cloud.ClusterPolicy.newBuilder().setType(Cloud.ClusterPolicy.PolicyType.ActiveStandby)
+ .setActiveClusterName("active")
+ .addStandbyClusterNames("standby")).build();
+ }
+}
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 194bc6a8a59..c74138023e0 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
@@ -36,6 +36,7 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -1062,6 +1063,49 @@ public class CloudSystemInfoServiceTest {
Assert.assertFalse(infoService.containsCloudCluster("cluster_2"));
}
+ @Test
+ public void testRemovingRecreatedComputeGroupKeepsCurrentNameMapping() {
+ infoService = new CloudSystemInfoService();
+ String clusterName = "recreated_cluster";
+ String oldClusterId = "old_cluster_id";
+ String newClusterId = "new_cluster_id";
+
+ Backend oldBackend = new Backend(1L, "127.0.0.1", 9050);
+ Map<String, String> oldTagMap = Tag.DEFAULT_BACKEND_TAG.toMap();
+ oldTagMap.put(Tag.CLOUD_CLUSTER_NAME, clusterName);
+ oldTagMap.put(Tag.CLOUD_CLUSTER_ID, oldClusterId);
+ oldBackend.setTagMap(oldTagMap);
+ infoService.updateCloudClusterMapNoLock(Arrays.asList(oldBackend), new ArrayList<>());
+
+ Backend newBackend = new Backend(2L, "127.0.0.2", 9050);
+ Map<String, String> newTagMap = Tag.DEFAULT_BACKEND_TAG.toMap();
+ newTagMap.put(Tag.CLOUD_CLUSTER_NAME, clusterName);
+ newTagMap.put(Tag.CLOUD_CLUSTER_ID, newClusterId);
+ newBackend.setTagMap(newTagMap);
+ infoService.updateCloudClusterMapNoLock(Arrays.asList(newBackend), new ArrayList<>());
+
+ infoService.updateCloudClusterMapNoLock(new ArrayList<>(), Arrays.asList(oldBackend));
+
+ Assert.assertEquals(newClusterId, infoService.getCloudClusterIdByName(clusterName));
+ Assert.assertNull(infoService.getComputeGroupById(oldClusterId));
+ Assert.assertNotNull(infoService.getComputeGroupById(newClusterId));
+ }
+
+ @Test
+ public void testRemovingCurrentComputeGroupRemovesNameMapping() {
+ infoService = new CloudSystemInfoService();
+ String clusterName = "removed_cluster";
+ String clusterId = "removed_cluster_id";
+ CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta(
+ clusterId, clusterName, CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE);
+ infoService.addComputeGroup(clusterId, computeGroup);
+
+ infoService.removeComputeGroup(clusterId, clusterName);
+
+ Assert.assertNull(infoService.getCloudClusterIdByName(clusterName));
+ Assert.assertNull(infoService.getComputeGroupById(clusterId));
+ }
+
/**
* Helper method to create a test ConnectContext with specific cluster name
*/
diff --git a/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out b/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out
new file mode 100644
index 00000000000..889a512b525
--- /dev/null
+++ b/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out
@@ -0,0 +1,21 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !recreated --
+1
+2
+3
+
+-- !renamed --
+1
+2
+3
+
+-- !reused --
+1
+2
+3
+
+-- !repaired --
+1
+2
+3
+
diff --git a/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy b/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy
new file mode 100644
index 00000000000..ac1e7b586f3
--- /dev/null
+++ b/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy
@@ -0,0 +1,169 @@
+// 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.
+
+import groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+
+// Run with a cloud image containing the checker debug points below. For a pre-fix
+// reproduction, keep the debug points but revert the reconciliation changes.
+suite('test_compute_group_name_reconciliation', 'cloud_p0,docker') {
+ if (!isCloudMode()) {
+ return
+ }
+
+ def options = new ClusterOptions()
+ options.cloudMode = true
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.enableDebugPoints()
+ options.feConfigs += ['cloud_cluster_check_interval_second=1', 'heartbeat_interval_second=1']
+
+ docker(options) {
+ def ms = cluster.getAllMetaservices().get(0)
+ def msEndpoint = "${ms.host}:${ms.httpPort}"
+ def feLog = new File(cluster.getFeByIndex(1).getLogFilePath())
+ def physicalPause = 'CloudClusterChecker.checkCloudBackends.pause'
+ def instancePause = 'CloudInstanceStatusChecker.runAfterCatalogReady.pause'
+ def stalePause = 'CloudInstanceStatusChecker.afterGetInstance.pause'
+ def debug = GetDebugPoint()
+
+ def alterCluster = { operation, group ->
+ httpTest {
+ endpoint msEndpoint
+ uri "/MetaService/http/${operation}?token=${token}"
+ body JsonOutput.toJson([instance_id: 'default_instance_id', cluster: group])
+ check { code, response ->
+ def result = parseJson(response)
+ assertTrue(code == 200 && result.code == 'OK', "${operation}: ${response}")
+ }
+ }
+ }
+ def remoteGroup = { name ->
+ def group = get_instance(ms).clusters.find { it.cluster_name == name }
+ assertNotNull(group, "remote compute group ${name}")
+ group
+ }
+ def pause = { point, phase, params = [:] ->
+ debug.enableDebugPointForAllFEs(point, params + [phase: phase, timeout: '180'])
+ // A unique phase proves this cycle reached the pause; no timing guesses.
+ awaitUntil(30) { feLog.text.contains("${point} phase=${phase}") }
+ }
+ def resume = { point -> debug.disableDebugPointForAllFEs(point) }
+ def waitPhysical = { name, id, removedIds ->
+ awaitUntil(60) {
+ def bes = sql_return_maparray('SHOW BACKENDS')
+ def tags = bes.collect { parseJson(it.Tag) }
+ def groups = sql_return_maparray('SHOW COMPUTE GROUPS')
+ tags.any { it.compute_group_id == id && it.compute_group_name == name } &&
+ !tags.any { removedIds.contains(it.compute_group_id) } &&
+ groups.any { it.Name == name && it.BackendNum.toInteger() == 1 }
+ }
+ sql "USE @${name}"
+ }
+
+ try {
+ cluster.addBackend(1, 'cg_reused')
+ cluster.addBackend(1, 'cg_spare')
+ def old = remoteGroup('cg_reused')
+ def spare = remoteGroup('cg_spare')
+ waitPhysical('cg_reused', old.cluster_id, [])
+ waitPhysical('cg_spare', spare.cluster_id, [])
+ sql 'USE @compute_cluster'
+ sql 'DROP TABLE IF EXISTS test_compute_group_name_reconciliation'
+ sql '''CREATE TABLE test_compute_group_name_reconciliation (k INT)
+ DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES ('replication_num'='1')'''
+ sql 'INSERT INTO test_compute_group_name_reconciliation VALUES (1), (2), (3)'
+
+ // 1. A single checker cycle sees a same-name replacement on distinct BEs.
+ pause(physicalPause, 'recreate')
+ alterCluster('drop_cluster', [cluster_id: old.cluster_id, cluster_name: 'cg_reused'])
+ alterCluster('drop_cluster', [cluster_id: spare.cluster_id, cluster_name: 'cg_spare'])
+ alterCluster('add_cluster', [cluster_id: 'cg_recreated_id', cluster_name: 'cg_reused',
+ type: 'COMPUTE', nodes: spare.nodes])
+ resume(physicalPause)
+ waitPhysical('cg_reused', 'cg_recreated_id', [old.cluster_id, spare.cluster_id])
+ order_qt_recreated 'SELECT k FROM test_compute_group_name_reconciliation'
+
+ // 2. Rename A and reuse its old name for B before the next checker cycle.
+ pause(physicalPause, 'rename')
+ alterCluster('rename_cluster', [cluster_id: 'cg_recreated_id', cluster_name: 'cg_renamed'])
+ alterCluster('add_cluster', [cluster_id: 'cg_reuse_after_rename_id', cluster_name: 'cg_reused',
+ type: 'COMPUTE', nodes: old.nodes])
+ resume(physicalPause)
+ waitPhysical('cg_renamed', 'cg_recreated_id', [old.cluster_id, spare.cluster_id])
+ order_qt_renamed 'SELECT k FROM test_compute_group_name_reconciliation'
+ waitPhysical('cg_reused', 'cg_reuse_after_rename_id', [old.cluster_id, spare.cluster_id])
+ order_qt_reused 'SELECT k FROM test_compute_group_name_reconciliation'
+
+ // 3. Hold an old virtual-group response while the physical checker installs
+ // a newer same-name physical group. Then let the old response overwrite it.
+ cluster.addBackend(1, 'cg_snapshot_spare')
+ def snapshotSpare = remoteGroup('cg_snapshot_spare')
+ waitPhysical('cg_snapshot_spare', snapshotSpare.cluster_id, [])
+ sql 'USE @compute_cluster'
+ pause(instancePause, 'before_virtual')
+ alterCluster('add_cluster', [cluster_id: 'cg_stale_virtual_id', cluster_name: 'cg_snapshot',
+ type: 'VIRTUAL', cluster_names: ['compute_cluster', 'cg_renamed'],
+ cluster_policy: [type: 'ActiveStandby', active_cluster_name: 'compute_cluster',
+ standby_cluster_names: ['cg_renamed']]])
+ debug.enableDebugPointForAllFEs(stalePause,
+ [cluster_id: 'cg_stale_virtual_id', phase: 'stale_response', timeout: '180'])
+ resume(instancePause)
+ awaitUntil(30) { feLog.text.contains("${stalePause} phase=stale_response") }
+ alterCluster('drop_cluster', [cluster_id: 'cg_stale_virtual_id', cluster_name: 'cg_snapshot'])
+ alterCluster('drop_cluster', [cluster_id: snapshotSpare.cluster_id, cluster_name: 'cg_snapshot_spare'])
+ alterCluster('add_cluster', [cluster_id: 'cg_current_physical_id', cluster_name: 'cg_snapshot',
+ type: 'COMPUTE', nodes: snapshotSpare.nodes])
+ waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id])
+ sql 'USE @compute_cluster'
+ pause(physicalPause, 'hold_repair')
+ debug.enableDebugPointForAllFEs(instancePause, [phase: 'after_stale', timeout: '180'])
+ resume(stalePause)
+ awaitUntil(30) { feLog.text.contains("${instancePause} phase=after_stale") }
+ awaitUntil(30) {
+ sql_return_maparray('SHOW COMPUTE GROUPS').any {
+ it.Name == 'cg_snapshot' && it.SubComputeGroups.contains('cg_renamed')
+ }
+ }
+
+ // Current instance data removes V. P's BEs remain, but its name is now missing.
+ resume(instancePause)
+ awaitUntil(60) {
+ !sql_return_maparray('SHOW COMPUTE GROUPS').any { it.Name == 'cg_snapshot' }
+ }
+ // No node or name change in MS: only periodic mapping repair can restore P.
+ resume(physicalPause)
+ waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id])
+ order_qt_repaired 'SELECT k FROM test_compute_group_name_reconciliation'
+
+ // Check stability after several subsequent physical AND instance cycles.
+ 3.times {
+ def logOffset = feLog.text.length()
+ awaitUntil(30) {
+ def subsequent = feLog.text.substring(logOffset)
+ subsequent.contains('daemon cluster get cluster info succ') &&
+ subsequent.contains('finished to cloud instance checker')
+ }
+ waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id])
+ }
+ } finally {
+ resume(stalePause)
+ resume(instancePause)
+ resume(physicalPause)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org