diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudComputeGroupMeta.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudComputeGroupMeta.java index 4d258ed262e391..23ce1a06b03dbe 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudComputeGroupMeta.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudComputeGroupMeta.java @@ -155,7 +155,7 @@ public Cloud.ClusterPolicy toPb() { @Getter @Setter - private Map properties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP); + private volatile Map properties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP); public CloudComputeGroupMeta(String id, String name, ComputeTypeEnum type) { this.id = id; @@ -245,7 +245,11 @@ public void checkProperties(Map inputProperties) throws DdlExcep } public void modifyProperties(Map inputProperties) throws DdlException { - String balanceType = inputProperties.get(BALANCE_TYPE); + applyBalanceTypeRule(properties, inputProperties.get(BALANCE_TYPE)); + } + + // only async_warmup carries a timeout: drop it for other types, fill the default when missing + private static void applyBalanceTypeRule(Map target, String balanceType) { if (balanceType == null) { return; } @@ -253,20 +257,20 @@ public void modifyProperties(Map inputProperties) throws DdlExce || BalanceTypeEnum.SYNC_WARMUP.getValue().equals(balanceType) || BalanceTypeEnum.PEER_READ_ASYNC_WARMUP.getValue().equals(balanceType)) { // delete BALANCE_WARM_UP_TASK_TIMEOUT if exists - properties.remove(BALANCE_WARM_UP_TASK_TIMEOUT); + target.remove(BALANCE_WARM_UP_TASK_TIMEOUT); } else if (BalanceTypeEnum.ASYNC_WARMUP.getValue().equals(balanceType)) { // if BALANCE_WARM_UP_TASK_TIMEOUT exists, it has been validated in validateProperty - if (!properties.containsKey(BALANCE_WARM_UP_TASK_TIMEOUT)) { - properties.put(BALANCE_WARM_UP_TASK_TIMEOUT, String.valueOf(DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT)); + if (!target.containsKey(BALANCE_WARM_UP_TASK_TIMEOUT)) { + target.put(BALANCE_WARM_UP_TASK_TIMEOUT, String.valueOf(DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT)); } } } - // set properties, just set in periodic instance status checker + // set properties, just set in periodic instance status checker. + // MS is the source of truth and stores only what was explicitly set, so FE properties = + // FE config defaults overlaid by the MS snapshot: keys removed in MS fall back to defaults. public void setProperties(Map propertiesInMs) { - if (propertiesInMs == null || propertiesInMs.isEmpty()) { - return; - } + Map newProperties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP); for (Map.Entry entry : propertiesInMs.entrySet()) { String key = entry.getKey(); @@ -281,9 +285,15 @@ public void setProperties(Map propertiesInMs) { } if (value != null && !value.isEmpty()) { - properties.put(key, value); + newProperties.put(key, value); } } + applyBalanceTypeRule(newProperties, newProperties.get(BALANCE_TYPE)); + + if (!newProperties.equals(properties)) { + LOG.info("compute group {} properties changed: {} -> {}", name, properties, newProperties); + properties = newProperties; + } } public BalanceTypeEnum getBalanceType() { 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..7cd5de07256564 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 @@ -121,48 +121,12 @@ private void handleComputeClusters(List computeClusters) { + "it may be wait cluster checker to sync, ignore it", computeClusterInMs); } else { - // exist compute group, check properties changed and update if needed - updatePropertiesIfChanged(computeGroupInFe, computeClusterInMs); + // exist compute group, resync properties from the authoritative MS snapshot + computeGroupInFe.setProperties(computeClusterInMs.getPropertiesMap()); } } } - /** - * Compare properties between compute cluster in MS and compute group in FE, - * update only the changed key-value pairs to avoid unnecessary updates. - */ - private void updatePropertiesIfChanged(CloudComputeGroupMeta computeGroupInFe, Cloud.ClusterPB computeClusterInMs) { - Map propertiesInMs = computeClusterInMs.getPropertiesMap(); - Map propertiesInFe = computeGroupInFe.getProperties(); - - if (propertiesInMs == null || propertiesInMs.isEmpty()) { - return; - } - Map changedProperties = new HashMap<>(); - - // Check for changed or new properties - for (Map.Entry entry : propertiesInMs.entrySet()) { - String key = entry.getKey(); - String valueInMs = entry.getValue(); - String valueInFe = propertiesInFe.get(key); - - if (valueInFe != null && valueInFe.equalsIgnoreCase(valueInMs)) { - continue; - } - changedProperties.put(key, valueInMs); - - LOG.debug("Property changed for compute group {}: {} = {} (was: {})", - computeGroupInFe.getName(), key, valueInMs, valueInFe); - } - - // Only update if there are actual changes - if (!changedProperties.isEmpty()) { - LOG.info("Updating properties for compute group {}: {}", - computeGroupInFe.getName(), changedProperties); - computeGroupInFe.setProperties(changedProperties); - } - } - private void categorizeClusters(List clusters, List virtualClusters, List computeClusters) { for (Cloud.ClusterPB cluster : clusters) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterComputeGroupCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterComputeGroupCommand.java index d5f520661b8f8e..4d237654d84e0f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterComputeGroupCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterComputeGroupCommand.java @@ -32,6 +32,7 @@ import org.apache.doris.qe.StmtExecutor; //import org.apache.commons.lang3.StringUtils; +import java.util.LinkedHashMap; import java.util.Map; /** @@ -87,8 +88,13 @@ public void validate(ConnectContext connectContext) throws UserException { public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception { validate(ctx); CloudSystemInfoService cloudSys = ((CloudSystemInfoService) Env.getCurrentSystemInfo()); + // MS replaces the whole property map, so send current (timeout rule already applied + // by validate) overlaid with the user input, not just the changed keys + Map merged = new LinkedHashMap<>( + cloudSys.getComputeGroupByName(computeGroupName).getProperties()); + merged.putAll(properties); // send rpc to ms - cloudSys.alterComputeGroupProperties(computeGroupName, properties); + cloudSys.alterComputeGroupProperties(computeGroupName, merged); } @Override 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 dd36c0f8b3bcbb..3535b84cc2b6e8 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 @@ -112,6 +112,32 @@ public void testSyncInstanceCreatesVirtualComputeGroup() { Assertions.assertEquals("standby_cg", virtualComputeGroup.getStandbyComputeGroup()); } + @Test + public void testSyncComputeGroupPropertiesRemovesKeysMissingFromMetaService() { + CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta( + "compute_cg_id", "compute_cg", CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE); + Map initialProperties = new HashMap<>(); + initialProperties.put(CloudComputeGroupMeta.BALANCE_TYPE, BalanceTypeEnum.ASYNC_WARMUP.getValue()); + initialProperties.put(CloudComputeGroupMeta.BALANCE_WARM_UP_TASK_TIMEOUT, "900"); + computeGroup.setProperties(initialProperties); + cloudSystemInfoService.addComputeGroup("compute_cg_id", computeGroup); + + Cloud.ClusterPB propertiesInMetaService = Cloud.ClusterPB.newBuilder() + .setClusterId("compute_cg_id") + .setClusterName("compute_cg") + .setType(Cloud.ClusterPB.Type.COMPUTE) + .putProperties(CloudComputeGroupMeta.BALANCE_TYPE, BalanceTypeEnum.WITHOUT_WARMUP.getValue()) + .build(); + Mockito.doReturn(instanceResponse(propertiesInMetaService)) + .when(cloudSystemInfoService).getCloudInstance(); + + new CloudInstanceStatusChecker(cloudSystemInfoService).runAfterCatalogReady(); + + Assertions.assertEquals(propertiesInMetaService.getPropertiesMap(), computeGroup.getProperties()); + Assertions.assertEquals(CloudComputeGroupMeta.DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT, + computeGroup.getBalanceWarmUpTaskTimeout()); + } + @Test public void testSyncInstanceCreatesVirtualComputeGroupAndCancelsTableLevelLoadEvent() throws Exception { databases.add(mockDb("ods", mockTable(1001, "orders"))); @@ -309,6 +335,19 @@ private Cloud.GetInstanceResponse instanceResponseWithoutVirtualComputeGroup() { .build(); } + private Cloud.GetInstanceResponse instanceResponse(Cloud.ClusterPB computeGroup) { + return Cloud.GetInstanceResponse.newBuilder() + .setStatus(Cloud.MetaServiceResponseStatus.newBuilder() + .setCode(Cloud.MetaServiceCode.OK) + .setMsg("OK") + .build()) + .setInstance(Cloud.InstanceInfoPB.newBuilder() + .setStatus(Cloud.InstanceInfoPB.Status.NORMAL) + .addClusters(computeGroup) + .build()) + .build(); + } + private Cloud.ClusterPB computeGroup(String computeGroupId, String computeGroupName) { return Cloud.ClusterPB.newBuilder() .setClusterId(computeGroupId)