Skip to content

[fix](cloud) Preserve recreated compute group mapping #67913

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deardeng wants to merge 4 commits into
base: master
Choose a base branch
from
Open

[fix](cloud) Preserve recreated compute group mapping #67913

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ca06f6e
- Test: Unit Test (RED confirmed: recreated-group case fails with exp…
deardeng Sep 13, 2026
11a4644
[fix](cloud) Reconcile compute group name ownership during synchroniz…
deardeng Sep 14, 2026
5da525f
[test](cloud) Skip name reconciliation case outside cloud mode
deardeng Sep 14, 2026
7e7c959
[fix](cloud) Exclude rejected virtual groups from name reconciliation
deardeng Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -146,19 +147,22 @@ private void checkToDelCluster(Map<String, ClusterPB> remoteClusterIdToPB, Set<S
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))) {
Comment thread
deardeng marked this conversation as resolved.
((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);
}
);
}
Expand Down Expand Up @@ -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<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.
Expand Down Expand Up @@ -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) {
Expand Down
59 changes: 45 additions & 14 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -58,13 +59,30 @@ 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)) {
return;
}

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());
Expand Down Expand Up @@ -107,8 +125,10 @@ private void processVirtualClusters(List<Cloud.ClusterPB> clusters) {
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) {
Expand Down Expand Up @@ -179,15 +199,23 @@ private void categorizeClusters(List<Cloud.ClusterPB> clusters,
}
}

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
Expand All @@ -201,6 +229,7 @@ private void handleVirtualClusters(List<Cloud.ClusterPB> virtualGroups, List<Clo
syncFileCacheTasksForVirtualGroup(virtualGroupInMs, virtualGroupInFe);
}
}
return reconciledGroups;
}

private void cancelCacheJobs(CloudComputeGroupMeta vcgInFe, List<String> jobIds) {
Expand Down Expand Up @@ -349,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) {
Expand Down Expand Up @@ -478,27 +508,27 @@ private void diffAndUpdateComputeGroup(Cloud.ClusterPB cluster, CloudComputeGrou
}
}

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(),
Expand All @@ -513,6 +543,7 @@ private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List<Cloud.Cl
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,
Expand Down
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).
*
Expand Down
74 changes: 45 additions & 29 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -247,17 +248,46 @@ 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);
}

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) {
Expand Down Expand Up @@ -421,11 +451,17 @@ 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);
// 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();
Expand Down Expand Up @@ -632,22 +668,13 @@ public void updateCloudClusterMapNoLock(List<Backend> toAdd, List<Backend> 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);
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);
Expand Down Expand Up @@ -1220,7 +1247,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();
Expand Down Expand Up @@ -1252,17 +1279,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<String> getCloudClusterNames() {
rlock.lock();
try {
Expand Down
Loading
Loading