From ca06f6eba8d3573e8d133edbab5856154bbe7792 Mon Sep 17 00:00:00 2001 From: deardeng Date: Sun, 13 Sep 2026 19:00:00 +0800 Subject: [PATCH 1/4] - Test: Unit Test (RED confirmed: recreated-group case fails with expected new_cluster_id but was null; current-group removal passes) - Behavior changed: No - Does this need documentation: No [fix](cloud) Preserve recreated compute group mapping Issue Number: None Related PR: None Problem Summary: Recreating a compute group with the same name and a new ID updates the name mapping to the new ID. Removing the obsolete group then unconditionally removed that name, leaving the new backends and compute group unreachable by name. Delete the mapping only when it still points to the group being removed, and remove duplicate cleanup from the backend deletion path. Fix compute group name resolution after recreating a group with the same name and a new ID. - Test: Unit Test (targeted CloudSystemInfoService tests, 2 passed) - Behavior changed: Yes (obsolete group cleanup preserves a same-name replacement mapping) - Does this need documentation: No --- .../cloud/system/CloudSystemInfoService.java | 10 +---- .../system/CloudSystemInfoServiceTest.java | 43 +++++++++++++++++++ 2 files changed, 44 insertions(+), 9 deletions(-) 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 b99e9967800be8..6863832453c761 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 @@ -250,7 +250,7 @@ public void addVirtualClusterInfoToMapsNoLock(String clusterId, String clusterNa 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); + clusterNameToId.remove(clusterName, clusterId); } public void renameVirtualClusterInfoFromMapsNoLock(String clusterId, String oldClusterName, String newClusterName) { @@ -637,17 +637,9 @@ public void updateCloudClusterMapNoLock(List toAdd, List toDel 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); 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 18734d8a1b0705..5962833c4c4da3 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 @@ -1062,6 +1062,49 @@ public void testContainsCloudCluster() { Assertions.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 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(List.of(oldBackend), new ArrayList<>()); + + Backend newBackend = new Backend(2L, "127.0.0.2", 9050); + Map 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(List.of(newBackend), new ArrayList<>()); + + infoService.updateCloudClusterMapNoLock(new ArrayList<>(), List.of(oldBackend)); + + Assertions.assertEquals(newClusterId, infoService.getCloudClusterIdByName(clusterName)); + Assertions.assertNull(infoService.getComputeGroupById(oldClusterId)); + Assertions.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); + + Assertions.assertNull(infoService.getCloudClusterIdByName(clusterName)); + Assertions.assertNull(infoService.getComputeGroupById(clusterId)); + } + /** * Helper method to create a test ConnectContext with specific cluster name */ From 11a46446995967f0eae5e2be046be6bbb7cacb34 Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 14 Sep 2026 14:56:03 +0800 Subject: [PATCH 2/4] [fix](cloud) Reconcile compute group name ownership during synchronization Related PR: #67913 Problem Summary: Adding a same-name replacement before deleting an obsolete group makes reverse name lookup fail and leaves old backends behind. Rename cleanup also removes names already owned by another group. Independently fetched physical and virtual snapshots can overwrite a current mapping, and unchanged nodes previously prevented later cycles from repairing it. Clean obsolete groups by ID using their own backend tags, guard both rename removals by the expected ID, and refresh name mappings for locally installed groups on each successful physical or virtual synchronization. This provides eventual repair after stale snapshots without adding snapshot versioning. Add a cloud Docker case with controlled checker pauses for recreation, rename and reuse, and a delayed virtual snapshot followed by periodic repair. Fix stale backend metadata and missing compute group name mappings after same-name recreation, rename and reuse, or delayed cloud metadata synchronization. - Test: Unit Test (GREEN checkpoint) - ./run-fe-ut.sh --run org.apache.doris.cloud.catalog.CloudClusterCheckerTest,org.apache.doris.cloud.catalog.CloudInstanceStatusCheckerTest,org.apache.doris.cloud.system.CloudSystemInfoServiceTest: 36 passed, 0 failures, 0 errors. - Targeted FE Checkstyle: 0 violations; git diff --check passed. - Cloud Docker regression added and Groovy syntax parsed; execution intentionally left to the user. No .out file generated or handwritten. - Behavior changed: Yes, periodic synchronization repairs name mappings and cleans obsolete group IDs even after name reuse. - Does this need documentation: No None --- .../cloud/catalog/CloudClusterChecker.java | 25 +- .../catalog/CloudInstanceStatusChecker.java | 19 ++ .../doris/cloud/catalog/CloudReplica.java | 2 +- .../cloud/system/CloudSystemInfoService.java | 63 +++-- .../catalog/CloudClusterCheckerTest.java | 236 ++++++++++++++++++ ...t_compute_group_name_reconciliation.groovy | 165 ++++++++++++ 6 files changed, 481 insertions(+), 29 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java create mode 100644 regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy 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 c8eee36f505a31..c38bc7debc81ae 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.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 @@ private void checkToDelCluster(Map remoteClusterIdToPB, Set 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 @@ private void checkCloudFes() { } private void checkCloudBackends() { + if (DebugPointUtil.isEnable("CloudClusterChecker.checkCloudBackends.pause")) { + LOG.info("CloudClusterChecker.checkCloudBackends.pause phase={}", DebugPointUtil.getDebugParamOrDefault( + "CloudClusterChecker.checkCloudBackends.pause", "phase", "")); + return; + } Map> 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 @@ private void checkCloudBackends() { // 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 cc9d256d566f38..bdbc53aa5ed4b9 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.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 CloudInstanceStatusChecker(CloudSystemInfoService cloudSystemInfoService) @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 @@ protected void runAfterCatalogReady() { } 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()); @@ -109,6 +127,7 @@ private void processVirtualClusters(List clusters) { handleComputeClusters(computeClusters); handleVirtualClusters(virtualClusters, computeClusters); removeObsoleteVirtualGroups(virtualClusters); + cloudSystemInfoService.refreshComputeGroupNames(virtualClusters); } private void handleComputeClusters(List computeClusters) { 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 b7ae1308ed5227..e06d413c908f29 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 @@ -628,7 +628,7 @@ public void clearClusterToBe(String cluster) { * * 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 6863832453c761..2e32d6816fa1c2 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 @@ -66,6 +66,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.HashSet; @@ -247,17 +248,45 @@ public void addVirtualClusterInfoToMapsNoLock(String clusterId, String clusterNa 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, clusterId); - } - 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. + // 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 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) { @@ -421,11 +450,15 @@ public String ownedByVirtualComputeGroup(String computeGroupName) { } } + // 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); + clusterNameToId.remove(computeGroupName, computeGroupId); invalidateCloudColocatePlacement(computeGroupId); } finally { wlock.unlock(); @@ -632,7 +665,6 @@ public void updateCloudClusterMapNoLock(List toAdd, List toDel 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); @@ -1212,7 +1244,7 @@ public void updateClusterNameToId(final String newName, final String originalName, final String clusterId) { wlock.lock(); try { - clusterNameToId.remove(originalName); + clusterNameToId.remove(originalName, clusterId); clusterNameToId.put(newName, clusterId); } finally { wlock.unlock(); @@ -1244,17 +1276,6 @@ public String getClusterNameByClusterIdNoLock(final String clusterId) { 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 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 00000000000000..d702e564cf9048 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java @@ -0,0 +1,236 @@ +// 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.List; +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 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 mockEnv() { + MockedStatic 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 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")); + } + + private void syncPhysical(Cloud.ClusterPB... groups) throws Exception { + Mockito.doReturn(Cloud.GetClusterResponse.newBuilder().setStatus(ok()) + .addAllCluster(List.of(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(List.of(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(List.of("active", "standby")) + .setClusterPolicy(Cloud.ClusterPolicy.newBuilder().setActiveClusterName("active") + .addStandbyClusterNames("standby")).build(); + } +} 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 00000000000000..183c178094f52b --- /dev/null +++ b/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy @@ -0,0 +1,165 @@ +// 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') { + 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) + } + } +} From 5da525f62f3d846165da9e870a6f044427681ab2 Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 14 Sep 2026 16:58:53 +0800 Subject: [PATCH 3/4] [test](cloud) Skip name reconciliation case outside cloud mode Related PR: #67913 Problem Summary: Skip the cloud-only compute group name reconciliation Docker suite before constructing the cluster when the regression environment is not in cloud mode. None - Test: Groovy syntax parsing and git diff --check passed. Docker case not run as requested. - Behavior changed: Yes, skip this regression suite in non-cloud mode. - Does this need documentation: No --- ...test_compute_group_name_reconciliation.out | 21 +++++++++++++++++++ ...t_compute_group_name_reconciliation.groovy | 4 ++++ 2 files changed, 25 insertions(+) create mode 100644 regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out 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 00000000000000..889a512b525aa4 --- /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 index 183c178094f52b..ac1e7b586f3d48 100644 --- 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 @@ -21,6 +21,10 @@ 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) From 7e7c959b8ad468f294f70a2f10cc68983f796d85 Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 14 Sep 2026 20:38:07 +0800 Subject: [PATCH 4/4] [fix](cloud) Exclude rejected virtual groups from name reconciliation ### What problem does this PR solve? Related PR: #67913 Problem Summary: A virtual-group rename with an invalid policy or subgroup shape is rejected by metadata reconciliation, but the raw snapshot previously still published its name. Removing that group later only removed its locally recorded name, leaving aliases pointing to a missing ID. Return reconciliation success for new and existing virtual groups and refresh names only for accepted records. Keep the complete snapshot for obsolete-group detection, and remove all name mappings owned by a removed ID under the existing write lock while preserving reused names. Add tests for rejected policy and subgroup updates, accepted renames, rejected new groups, and stale-alias cleanup. Explicitly set the valid policy type in test snapshots. Warm-up job ownership by compute-group ID remains out of scope. ### Release note Prevent rejected virtual-group updates from publishing invalid name mappings and clean up stale aliases when compute groups are removed. ### Check List (For Author) - Test: Unit Test: bash run-fe-ut.sh --run org.apache.doris.cloud.catalog.CloudClusterCheckerTest,org.apache.doris.cloud.catalog.CloudInstanceStatusCheckerTest,org.apache.doris.cloud.system.CloudSystemInfoServiceTest; 41 tests passed, 0 failures, 0 errors, 0 skipped. Before the fix, CloudClusterCheckerTest reproduced 3 intended assertion failures out of 11 tests. FE Checkstyle passed with 0 violations; git diff --check passed. Tests were not rerun for this history-only squash; the resulting tree is unchanged. Docker regression was not run, as requested; no regression output was changed. - Behavior changed: Yes (rejected virtual updates no longer publish names or proceed to cache-task synchronization; deletion clears all aliases owned by the removed ID) - Does this need documentation: No --- .../catalog/CloudInstanceStatusChecker.java | 42 +++++++---- .../cloud/system/CloudSystemInfoService.java | 5 +- .../catalog/CloudClusterCheckerTest.java | 74 ++++++++++++++++++- 3 files changed, 104 insertions(+), 17 deletions(-) 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 bdbc53aa5ed4b9..72742911dafbd4 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 @@ -125,9 +125,10 @@ private void processVirtualClusters(List clusters) { List computeClusters = new ArrayList<>(); categorizeClusters(clusters, virtualClusters, computeClusters); handleComputeClusters(computeClusters); - handleVirtualClusters(virtualClusters, computeClusters); + List 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(virtualClusters); + cloudSystemInfoService.refreshComputeGroupNames(reconciledVirtualClusters); } private void handleComputeClusters(List computeClusters) { @@ -198,15 +199,23 @@ private void categorizeClusters(List clusters, } } - private void handleVirtualClusters(List virtualGroups, List computeClusters) { + private List handleVirtualClusters( + List virtualGroups, List computeClusters) { + List 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 @@ -220,6 +229,7 @@ private void handleVirtualClusters(List virtualGroups, List jobIds) { @@ -368,21 +378,22 @@ private void syncFileCacheTasksForVirtualGroup( } } - 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) { @@ -497,27 +508,27 @@ private void diffAndUpdateComputeGroup(Cloud.ClusterPB cluster, CloudComputeGrou } } - private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List computeClusters) { + private boolean handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List computeClusters) { List 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(), @@ -532,6 +543,7 @@ private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List subClusterNames, Cloud.ClusterPB cluster, 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 2e32d6816fa1c2..b0f334d354dc31 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 @@ -265,6 +265,7 @@ public void renameVirtualClusterInfoFromMapsNoLock(String clusterId, String oldC // 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 remoteComputeGroups) { @@ -458,7 +459,9 @@ public void removeComputeGroup(String computeGroupId, String computeGroupName) { LOG.info("remove compute group, id={}, name={}", computeGroupId, computeGroupName); computeGroupIdToComputeGroup.remove(computeGroupId); clusterIdToBackend.remove(computeGroupId); - clusterNameToId.remove(computeGroupName, 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(); 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 index d702e564cf9048..bd20866e1598f1 100644 --- 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 @@ -196,6 +196,77 @@ void testLaterCyclesRepairMissingPhysicalAndVirtualNames() throws Exception { 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(List.of(groups)).build()).when(service).getCloudCluster("", "", ""); @@ -230,7 +301,8 @@ private Cloud.ClusterPB physical(String id, String name, String ip) { private Cloud.ClusterPB virtual(String id, String name) { return Cloud.ClusterPB.newBuilder().setClusterId(id).setClusterName(name) .setType(Cloud.ClusterPB.Type.VIRTUAL).addAllClusterNames(List.of("active", "standby")) - .setClusterPolicy(Cloud.ClusterPolicy.newBuilder().setActiveClusterName("active") + .setClusterPolicy(Cloud.ClusterPolicy.newBuilder().setType(Cloud.ClusterPolicy.PolicyType.ActiveStandby) + .setActiveClusterName("active") .addStandbyClusterNames("standby")).build(); } }