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]
<41898282+github-actions[bot]@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 <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
