This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5839326781b [fix](cloud) Fix VCG failed sync state convergence (#68833)
5839326781b is described below
commit 5839326781b8f5f3a668b50e9e3364dccc14a38e
Author: deardeng <[email protected]>
AuthorDate: Fri Oct 30 10:29:22 2026 +0800
[fix](cloud) Fix VCG failed sync state convergence (#68833)
Problem Summary: The FE cloud instance checker updated failed
synchronization timestamps per sub-group and only validated newly
created virtual compute groups. A healthy sibling could clear a real
failure, existing groups could acquire missing sub-groups without being
tracked, and drop or rename left stale timestamps. Evaluate the complete
sub-group set on every synchronization cycle and clear state when all
references recover or the virtual group leaves its old lifecycle
identity.
---
.../cloud/catalog/CloudInstanceStatusChecker.java | 73 ++++++++----
.../catalog/CloudInstanceStatusCheckerTest.java | 132 ++++++++++++++++++++-
2 files changed, 179 insertions(+), 26 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 762628f0f2e..bb4cbf86cf4 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
@@ -48,7 +48,7 @@ import java.util.stream.Collectors;
public class CloudInstanceStatusChecker extends MasterDaemon {
private static final Logger LOG =
LogManager.getLogger(CloudInstanceStatusChecker.class);
private CloudSystemInfoService cloudSystemInfoService;
- // if find vcg failed sync, record it timestamp, virtual compute group
name <-> timestamp
+ // if find vcg failed sync, record it timestamp, virtual compute group id
<-> timestamp
private Map<String, Long> lastFailedSyncTimeMap = new
ConcurrentHashMap<>();
public CloudInstanceStatusChecker(CloudSystemInfoService
cloudSystemInfoService) {
@@ -125,7 +125,11 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
List<Cloud.ClusterPB> computeClusters = new ArrayList<>();
categorizeClusters(clusters, virtualClusters, computeClusters);
handleComputeClusters(computeClusters);
- List<Cloud.ClusterPB> reconciledVirtualClusters =
handleVirtualClusters(virtualClusters, computeClusters);
+ Map<String, Cloud.ClusterPB> computeClustersByName =
computeClusters.stream()
+ .collect(Collectors.toMap(Cloud.ClusterPB::getClusterName,
computeCluster -> computeCluster));
+ Map<String, String> computeClusterIdsByNameInFe =
cloudSystemInfoService.getCloudClusterNameToId(false);
+ List<Cloud.ClusterPB> reconciledVirtualClusters =
+ handleVirtualClusters(virtualClusters, computeClustersByName,
computeClusterIdsByNameInFe);
// A rejected update still proves that the group exists in MS, so use
the full snapshot for removal.
removeObsoleteVirtualGroups(virtualClusters);
cloudSystemInfoService.refreshComputeGroupNames(reconciledVirtualClusters);
@@ -163,17 +167,20 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
}
- private List<Cloud.ClusterPB> handleVirtualClusters(
- List<Cloud.ClusterPB> virtualGroups, List<Cloud.ClusterPB>
computeClusters) {
+ private List<Cloud.ClusterPB> handleVirtualClusters(List<Cloud.ClusterPB>
virtualGroups,
+ Map<String, Cloud.ClusterPB>
computeClustersByName,
+ Map<String, String>
computeClusterIdsByNameInFe) {
List<Cloud.ClusterPB> reconciledGroups = new ArrayList<>();
for (Cloud.ClusterPB virtualGroupInMs : virtualGroups) {
CloudComputeGroupMeta virtualGroupInFe = cloudSystemInfoService
.getComputeGroupById(virtualGroupInMs.getClusterId());
boolean reconciled;
if (virtualGroupInFe != null) {
- reconciled =
handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe);
+ reconciled =
handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe,
+ computeClustersByName, computeClusterIdsByNameInFe);
} else {
- reconciled = handleNewVirtualComputeGroup(virtualGroupInMs,
computeClusters);
+ reconciled = handleNewVirtualComputeGroup(virtualGroupInMs,
+ computeClustersByName, computeClusterIdsByNameInFe);
}
// Rejected renames must not publish a name for metadata whose
update was rejected.
if (!reconciled) {
@@ -343,7 +350,9 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
private boolean handleExistingVirtualComputeGroup(
- Cloud.ClusterPB clusterInMs, CloudComputeGroupMeta
virtualGroupInFe) {
+ Cloud.ClusterPB clusterInMs, CloudComputeGroupMeta
virtualGroupInFe,
+ Map<String, Cloud.ClusterPB> computeClustersByName,
+ Map<String, String> computeClusterIdsByNameInFe) {
if (!isClusterIdConsistent(clusterInMs, virtualGroupInFe)) {
return false;
}
@@ -356,6 +365,8 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
return false;
}
+ checkSubClusters(clusterInMs.getClusterNamesList(), clusterInMs,
+ computeClustersByName, computeClusterIdsByNameInFe);
diffAndUpdateComputeGroup(clusterInMs, virtualGroupInFe);
return true;
}
@@ -472,7 +483,9 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
}
- private boolean handleNewVirtualComputeGroup(Cloud.ClusterPB cluster,
List<Cloud.ClusterPB> computeClusters) {
+ private boolean handleNewVirtualComputeGroup(
+ Cloud.ClusterPB cluster, Map<String, Cloud.ClusterPB>
computeClustersByName,
+ Map<String, String> computeClusterIdsByNameInFe) {
List<String> subComputeGroups = cluster.getClusterNamesList();
if (subComputeGroups.isEmpty()) {
LOG.info("found virtual cluster {} which has no sub clusters, skip
empty virtual cluster", cluster);
@@ -494,7 +507,7 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
LOG.warn("virtual compute err, standby cluster size not eq 1 in ms
{}", cluster);
return false;
}
- checkSubClusters(subComputeGroups, cluster, computeClusters);
+ checkSubClusters(subComputeGroups, cluster, computeClustersByName,
computeClusterIdsByNameInFe);
CloudComputeGroupMeta computeGroup = new
CloudComputeGroupMeta(cluster.getClusterId(),
cluster.getClusterName(),
CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL);
computeGroup.setSubComputeGroups(new ArrayList<>(subComputeGroups));
@@ -511,32 +524,45 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
private void checkSubClusters(List<String> subClusterNames,
Cloud.ClusterPB cluster,
- List<Cloud.ClusterPB> computeClustersInPB) {
+ Map<String, Cloud.ClusterPB>
computeClustersByName,
+ Map<String, String>
computeClusterIdsByNameInFe) {
+ boolean allSubClustersExist = true;
for (String subClusterName : subClusterNames) {
- if (cloudSystemInfoService.getCloudClusterIdByName(subClusterName)
== null) {
- handleFailedSync(cluster, subClusterName, computeClustersInPB);
- continue;
+ Cloud.ClusterPB subClusterInMs =
computeClustersByName.get(subClusterName);
+ String subClusterIdInFe =
computeClusterIdsByNameInFe.get(subClusterName);
+ boolean sameSubClusterInFe = subClusterInMs != null
+ && subClusterInMs.getClusterId().equals(subClusterIdInFe);
+ // Empty compute groups exist only in MS because FE builds its
mapping from BE nodes.
+ boolean emptySubClusterOnlyInMs = subClusterInMs != null &&
subClusterInMs.getNodesCount() == 0
+ && subClusterIdInFe == null;
+ if (!sameSubClusterInFe && !emptySubClusterOnlyInMs) {
+ allSubClustersExist = false;
+ handleFailedSync(cluster, subClusterName, subClusterInMs,
subClusterIdInFe);
}
- // CloudClusterChecker find sub compute group
- lastFailedSyncTimeMap.remove(cluster.getClusterName());
+ }
+ if (allSubClustersExist) {
+ lastFailedSyncTimeMap.remove(cluster.getClusterId());
}
}
private void handleFailedSync(Cloud.ClusterPB cluster, String
subClusterName,
- List<Cloud.ClusterPB> computeClustersInPB) {
- if (!lastFailedSyncTimeMap.containsKey(cluster.getClusterName())) {
- lastFailedSyncTimeMap.put(cluster.getClusterName(),
System.currentTimeMillis());
+ Cloud.ClusterPB subClusterInMs, String
subClusterIdInFe) {
+ if (!lastFailedSyncTimeMap.containsKey(cluster.getClusterId())) {
+ lastFailedSyncTimeMap.put(cluster.getClusterId(),
System.currentTimeMillis());
} else {
- List<String> computeGroupsInPb = computeClustersInPB.stream()
-
.map(Cloud.ClusterPB::getClusterName).collect(Collectors.toList());
- if (computeGroupsInPb.contains(subClusterName)) {
+ if (subClusterInMs == null) {
+ LOG.warn("ms cant find {}, current fe cluster id: {}, it may
be dropped or renamed",
+ subClusterName, subClusterIdInFe);
+ } else if (subClusterIdInFe == null) {
LOG.warn("fe mem cant find {}, it may be wait cluster check to
sync", subClusterName);
} else {
- LOG.warn("fe mem and ms cant find {}, it may be dropped or
renamed", subClusterName);
+ LOG.warn("sub cluster id mismatch for {}, fe id: {}, ms id:
{}, "
+ + "it may be wait cluster check to sync",
+ subClusterName, subClusterIdInFe,
subClusterInMs.getClusterId());
}
// sub cluster may be dropped or rename, or fe may be slowly,
// need manual intervention
- if (System.currentTimeMillis() -
lastFailedSyncTimeMap.get(cluster.getClusterName())
+ if (System.currentTimeMillis() -
lastFailedSyncTimeMap.get(cluster.getClusterId())
> 3 * Config.cloud_cluster_check_interval_second * 1000L) {
LOG.warn("virtual compute err, cant find cluster info by
cluster checker, "
+ "sub cluster: {}, virtual cluster: {}",
subClusterName, cluster.getClusterName());
@@ -551,6 +577,7 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
// in fe mem, but not in meta server
if (!msVirtualClusters.contains(computeGroup.getId())) {
LOG.info("virtual compute group {} will be removed.",
computeGroup.getName());
+ lastFailedSyncTimeMap.remove(computeGroup.getId());
MetricRepo.unregisterCloudMetrics(computeGroup.getId(),
computeGroup.getName(),
Collections.emptyList());
cloudSystemInfoService.removeComputeGroup(computeGroup.getId(),
computeGroup.getName());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
index 3535b84cc2b..81cbaa687e8 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
@@ -28,6 +28,7 @@ import
org.apache.doris.cloud.OnTablesFilter.TableFilterRule.RuleType;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
+import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.nereids.trees.plans.commands.WarmUpClusterCommand;
import org.apache.doris.persist.EditLog;
@@ -138,6 +139,100 @@ public class CloudInstanceStatusCheckerTest {
computeGroup.getBalanceWarmUpTaskTimeout());
}
+ @Test
+ public void testVirtualComputeGroupKeepsFailureUntilAllSubGroupsExist() {
+ addComputeGroup("standby_cg_id", "standby_cg");
+
Mockito.doReturn(instanceResponseWithVirtualComputeGroup()).when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+
+ Assertions.assertTrue(failedSyncTimes(checker).containsKey("vcg_id"));
+ }
+
+ @Test
+ public void testVirtualComputeGroupAcceptsEmptySubGroups() {
+
Mockito.doReturn(instanceResponseWithEmptyComputeGroups()).when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+
+ Assertions.assertTrue(failedSyncTimes(checker).isEmpty());
+ }
+
+ @Test
+ public void testVirtualComputeGroupRejectsMismatchedFeSubGroupId() {
+ addComputeGroup("wrong_active_cg_id", "active_cg");
+ addComputeGroup("standby_cg_id", "standby_cg");
+
Mockito.doReturn(instanceResponseWithVirtualComputeGroup()).when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+
+ Assertions.assertTrue(failedSyncTimes(checker).containsKey("vcg_id"));
+ }
+
+ @Test
+ public void
testExistingVirtualComputeGroupDoesNotRecoverFromFeOnlySubGroup() {
+ addComputeGroup("active_cg_id", "active_cg");
+ addComputeGroup("standby_cg_id", "standby_cg");
+ Cloud.GetInstanceResponse missingActiveResponse =
+ instanceResponseWithVirtualComputeGroup("missing_active_cg",
"standby_cg");
+ Cloud.GetInstanceResponse recoveredActiveResponse =
instanceResponseWithVirtualComputeGroup(
+ "vcg", "missing_active_cg", "standby_cg",
+ computeGroup("missing_active_cg_id", "missing_active_cg"),
+ computeGroup("standby_cg_id", "standby_cg"));
+ Mockito.doReturn(instanceResponseWithVirtualComputeGroup())
+ .doReturn(missingActiveResponse)
+ .doReturn(missingActiveResponse)
+ .doReturn(recoveredActiveResponse)
+ .when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+ Assertions.assertTrue(failedSyncTimes(checker).isEmpty());
+
+ checker.runAfterCatalogReady();
+ Assertions.assertTrue(failedSyncTimes(checker).containsKey("vcg_id"));
+
+ addComputeGroup("missing_active_cg_id", "missing_active_cg");
+ checker.runAfterCatalogReady();
+ Assertions.assertTrue(failedSyncTimes(checker).containsKey("vcg_id"));
+
+ checker.runAfterCatalogReady();
+ Assertions.assertTrue(failedSyncTimes(checker).isEmpty());
+ }
+
+ @Test
+ public void testDropInvalidVirtualComputeGroupClearsFailure() {
+ Mockito.doReturn(instanceResponseWithVirtualComputeGroup())
+ .doReturn(instanceResponseWithoutVirtualComputeGroup())
+ .when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+ Assertions.assertTrue(failedSyncTimes(checker).containsKey("vcg_id"));
+
+ checker.runAfterCatalogReady();
+
Assertions.assertNull(cloudSystemInfoService.getComputeGroupById("vcg_id"));
+ Assertions.assertTrue(failedSyncTimes(checker).isEmpty());
+ }
+
+ @Test
+ public void testRenameVirtualComputeGroupKeepsFailureTime() {
+ Mockito.doReturn(instanceResponseWithVirtualComputeGroup("old_vcg",
"active_cg", "standby_cg"))
+ .doReturn(instanceResponseWithVirtualComputeGroup("new_vcg",
"active_cg", "standby_cg"))
+ .when(cloudSystemInfoService).getCloudInstance();
+
+ CloudInstanceStatusChecker checker = new
CloudInstanceStatusChecker(cloudSystemInfoService);
+ checker.runAfterCatalogReady();
+ long failedSyncTime = 1L;
+ failedSyncTimes(checker).put("vcg_id", failedSyncTime);
+
+ checker.runAfterCatalogReady();
+ Assertions.assertEquals(failedSyncTime,
failedSyncTimes(checker).get("vcg_id"));
+ }
+
@Test
public void
testSyncInstanceCreatesVirtualComputeGroupAndCancelsTableLevelLoadEvent()
throws Exception {
databases.add(mockDb("ods", mockTable(1001, "orders")));
@@ -291,14 +386,32 @@ public class CloudInstanceStatusCheckerTest {
}
private Cloud.GetInstanceResponse
instanceResponseWithVirtualComputeGroup(String active, String standby) {
+ return instanceResponseWithVirtualComputeGroup("vcg", active, standby);
+ }
+
+ private Cloud.GetInstanceResponse instanceResponseWithVirtualComputeGroup(
+ String virtualComputeGroupName, String active, String standby) {
Cloud.ClusterPB activeComputeGroup = computeGroup("active_cg_id",
"active_cg");
Cloud.ClusterPB standbyComputeGroup = computeGroup("standby_cg_id",
"standby_cg");
+ return instanceResponseWithVirtualComputeGroup(
+ virtualComputeGroupName, active, standby, activeComputeGroup,
standbyComputeGroup);
+ }
+
+ private Cloud.GetInstanceResponse instanceResponseWithEmptyComputeGroups()
{
+ return instanceResponseWithVirtualComputeGroup("vcg", "active_cg",
"standby_cg",
+ emptyComputeGroup("active_cg_id", "active_cg"),
+ emptyComputeGroup("standby_cg_id", "standby_cg"));
+ }
+
+ private Cloud.GetInstanceResponse instanceResponseWithVirtualComputeGroup(
+ String virtualComputeGroupName, String active, String standby,
+ Cloud.ClusterPB activeComputeGroup, Cloud.ClusterPB
standbyComputeGroup) {
Cloud.ClusterPB virtualComputeGroup = Cloud.ClusterPB.newBuilder()
.setClusterId("vcg_id")
- .setClusterName("vcg")
+ .setClusterName(virtualComputeGroupName)
.setType(Cloud.ClusterPB.Type.VIRTUAL)
- .addClusterNames("active_cg")
- .addClusterNames("standby_cg")
+ .addClusterNames(active)
+ .addClusterNames(standby)
.setClusterPolicy(Cloud.ClusterPolicy.newBuilder()
.setType(Cloud.ClusterPolicy.PolicyType.ActiveStandby)
.setActiveClusterName(active)
@@ -319,6 +432,10 @@ public class CloudInstanceStatusCheckerTest {
.build();
}
+ private Map<String, Long> failedSyncTimes(CloudInstanceStatusChecker
checker) {
+ return Deencapsulation.getField(checker, "lastFailedSyncTimeMap");
+ }
+
private Cloud.GetInstanceResponse
instanceResponseWithoutVirtualComputeGroup() {
Cloud.ClusterPB activeComputeGroup = computeGroup("active_cg_id",
"active_cg");
Cloud.ClusterPB standbyComputeGroup = computeGroup("standby_cg_id",
"standby_cg");
@@ -349,6 +466,15 @@ public class CloudInstanceStatusCheckerTest {
}
private Cloud.ClusterPB computeGroup(String computeGroupId, String
computeGroupName) {
+ return Cloud.ClusterPB.newBuilder()
+ .setClusterId(computeGroupId)
+ .setClusterName(computeGroupName)
+ .setType(Cloud.ClusterPB.Type.COMPUTE)
+ .addNodes(Cloud.NodeInfoPB.getDefaultInstance())
+ .build();
+ }
+
+ private Cloud.ClusterPB emptyComputeGroup(String computeGroupId, String
computeGroupName) {
return Cloud.ClusterPB.newBuilder()
.setClusterId(computeGroupId)
.setClusterName(computeGroupName)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
