wuchong commented on code in PR #3614: URL: https://github.com/apache/fluss/pull/3614#discussion_r3564253529
########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/KvLeaderReplicaCapacityManager.java: ########## @@ -0,0 +1,221 @@ +/* + * 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.fluss.server.coordinator; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.cluster.ServerReconfigurable; +import org.apache.fluss.exception.ConfigException; +import org.apache.fluss.exception.DatabaseNotExistException; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.metadata.CoordinatorMetadataCache; +import org.apache.fluss.server.metadata.ServerInfo; + +import java.util.List; +import java.util.Set; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Manages the in-memory cluster capacity for KV leader replicas. */ +public class KvLeaderReplicaCapacityManager implements ServerReconfigurable { + + /** Negative capacity means the automatic capacity limit is disabled. */ + public static final long CAPACITY_LIMIT_DISABLED = -1L; + + private final Object lock = new Object(); + private final CoordinatorMetadataCache metadataCache; + + private long leaderReplicaMemoryReservedBytes; + private long currentKvLeaderReplicaCount; + + public KvLeaderReplicaCapacityManager( + Configuration conf, CoordinatorMetadataCache metadataCache) { + this.metadataCache = checkNotNull(metadataCache, "metadataCache should not be null."); + this.leaderReplicaMemoryReservedBytes = getLeaderReplicaMemoryReservedBytes(conf); + } + + @Override + public void validate(Configuration newConfig) throws ConfigException { + long memoryReservedBytes = getLeaderReplicaMemoryReservedBytes(newConfig); + if (memoryReservedBytes <= 0) { + throw new ConfigException( + ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED.key() + + " must be greater than 0."); + } + } + + @Override + public void reconfigure(Configuration newConfig) throws ConfigException { + long memoryReservedBytes = getLeaderReplicaMemoryReservedBytes(newConfig); + synchronized (lock) { + leaderReplicaMemoryReservedBytes = memoryReservedBytes; + } + } + + /** + * Checks whether the requested replicas fit the current capacity and reserves them if they do. + */ + public void checkAndIncrease(long newKvLeaderReplicaCount) { + if (newKvLeaderReplicaCount <= 0) { + return; + } + + synchronized (lock) { + long capacity = getKvLeaderReplicaCapacityLocked(); + long newTotal = currentKvLeaderReplicaCount + newKvLeaderReplicaCount; + if (capacity != CAPACITY_LIMIT_DISABLED && newTotal > capacity) { + throw new InsufficientKvLeaderReplicaCapacityException( + String.format( + "Not enough KV leader replica capacity. " + + "currentKvLeaderReplicaCount=%s, " + + "newKvLeaderReplicaCount=%s, " + + "kvLeaderReplicaCapacity=%s.", + currentKvLeaderReplicaCount, newKvLeaderReplicaCount, capacity)); + } + + currentKvLeaderReplicaCount = newTotal; + } + } + + /** Releases replicas that were removed from metadata or reserved by a failed creation. */ + public void decrease(long removedKvLeaderReplicaCount) { + if (removedKvLeaderReplicaCount <= 0) { + return; + } + + synchronized (lock) { + currentKvLeaderReplicaCount = + Math.max(0, currentKvLeaderReplicaCount - removedKvLeaderReplicaCount); + } + } + + /** Rebuilds the current KV leader replica count from metadata. */ + public void rebuildCurrentCount(MetadataManager metadataManager) { + long rebuiltCount = 0; + List<String> databases = metadataManager.listDatabases(); + for (String database : databases) { + List<String> tables; + try { + tables = metadataManager.listTables(database); + } catch (DatabaseNotExistException e) { + continue; + } catch (FlussRuntimeException e) { + if (!metadataManager.databaseExists(database)) { + continue; + } + throw e; + } + for (String table : tables) { + TablePath tablePath = TablePath.of(database, table); + TableInfo tableInfo; + try { + tableInfo = metadataManager.getTable(tablePath); + } catch (TableNotExistException e) { + continue; + } catch (FlussRuntimeException e) { + if (!metadataManager.tableExists(tablePath)) { + continue; + } + throw e; + } + rebuiltCount += getKvLeaderReplicaCount(tableInfo, metadataManager); + } + } + + synchronized (lock) { + currentKvLeaderReplicaCount = rebuiltCount; + } + } + + /** Returns the current in-memory KV leader replica count. */ + public long getKvLeaderReplicaCount() { + synchronized (lock) { + return currentKvLeaderReplicaCount; + } + } + + /** Returns the current cluster KV leader replica capacity, or -1 if disabled. */ + public long getKvLeaderReplicaCapacity() { + synchronized (lock) { + return getKvLeaderReplicaCapacityLocked(); + } + } + + @VisibleForTesting + long getLeaderReplicaMemoryReservedBytes() { + synchronized (lock) { + return leaderReplicaMemoryReservedBytes; + } + } + + private long getKvLeaderReplicaCapacityLocked() { + Set<ServerInfo> liveTabletServers = metadataCache.getLiveTabletServerInfos(); + int liveTabletServerCount = liveTabletServers.size(); + if (liveTabletServerCount == 0) { + return CAPACITY_LIMIT_DISABLED; + } + + int knownMemoryTabletServerCount = 0; + long knownMemoryBytes = 0; + for (ServerInfo serverInfo : liveTabletServers) { + if (serverInfo.resource().hasMemory()) { + knownMemoryTabletServerCount++; + knownMemoryBytes += serverInfo.resource().getMemoryBytes(); + } + } + + if (knownMemoryTabletServerCount * 2 <= liveTabletServerCount) { + return CAPACITY_LIMIT_DISABLED; + } + + long averageKnownMemoryBytes = knownMemoryBytes / knownMemoryTabletServerCount; + long totalMemoryBytes = + knownMemoryBytes + + averageKnownMemoryBytes + * (liveTabletServerCount - knownMemoryTabletServerCount); + return totalMemoryBytes / leaderReplicaMemoryReservedBytes; Review Comment: Add debug log for the capacity calculation details. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java: ########## @@ -240,20 +244,28 @@ protected void initCoordinatorStandby() throws Exception { this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true); this.remoteDirDynamicLoader = new RemoteDirDynamicLoader(conf); + this.metadataCache = new CoordinatorMetadataCache(); + this.kvLeaderReplicaCapacityManager = + new KvLeaderReplicaCapacityManager(conf, metadataCache); + serverMetricGroup.gauge( + MetricNames.KV_LEADER_REPLICA_COUNT, + kvLeaderReplicaCapacityManager::getKvLeaderReplicaCount); + serverMetricGroup.gauge( + MetricNames.KV_LEADER_REPLICA_CAPACITY, + kvLeaderReplicaCapacityManager::getKvLeaderReplicaCapacity); Review Comment: Move the metric register into `KvLeaderReplicaCapacityManager`. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java: ########## Review Comment: Restrict the number of constructor. Esp. for the constructor only used for tests, mark them package-visible and `@VisibleForTesting`. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/KvLeaderReplicaCapacityManager.java: ########## @@ -0,0 +1,221 @@ +/* + * 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.fluss.server.coordinator; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.cluster.ServerReconfigurable; +import org.apache.fluss.exception.ConfigException; +import org.apache.fluss.exception.DatabaseNotExistException; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.metadata.CoordinatorMetadataCache; +import org.apache.fluss.server.metadata.ServerInfo; + +import java.util.List; +import java.util.Set; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Manages the in-memory cluster capacity for KV leader replicas. */ +public class KvLeaderReplicaCapacityManager implements ServerReconfigurable { + + /** Negative capacity means the automatic capacity limit is disabled. */ + public static final long CAPACITY_LIMIT_DISABLED = -1L; + + private final Object lock = new Object(); + private final CoordinatorMetadataCache metadataCache; + + private long leaderReplicaMemoryReservedBytes; + private long currentKvLeaderReplicaCount; + + public KvLeaderReplicaCapacityManager( + Configuration conf, CoordinatorMetadataCache metadataCache) { + this.metadataCache = checkNotNull(metadataCache, "metadataCache should not be null."); + this.leaderReplicaMemoryReservedBytes = getLeaderReplicaMemoryReservedBytes(conf); + } Review Comment: `kv.leader-replica.memory-reserved` currently accepts `0b` from the static configuration, but the capacity calculation divides the reported TabletServer memory by this value and can therefore throw `ArithmeticException`. Dynamic validation is also inconsistent because it rejects zero. I think we should allow defining `0` as the explicit switch for disabling memory-based KV leader replica capacity control. Map it to `CAPACITY_LIMIT_DISABLED` before performing the division, allow it in dynamic validation, and document the behavior in the config description. Positive values should enable capacity estimation as usual, while negative values remain invalid. The improved option description: ```java "The estimated memory consumption of each KV leader replica, " + "used by the CoordinatorServer to calculate the cluster-level " + "KV leader replica capacity. This value does not reserve or " + "enforce memory on the tablet server. A value of 0 disables " + "memory-based KV leader replica capacity control."); ``` ########## fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java: ########## @@ -547,6 +555,22 @@ public class ConfigOptions { "The rack for the tabletServer. This will be used in rack aware bucket assignment " + "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`"); + public static final ConfigOption<Double> TABLET_SERVER_RESOURCE_CPU_CORES = + key("tablet-server.resource.cpu-cores") Review Comment: **Clarify that TabletServer resource options are advertised capacity, not enforced limits** The current keys, `tablet-server.resource.cpu-cores` and `tablet-server.resource.memory-size`, sound like they configure or enforce the resources available to the TabletServer process, such as a CPU quota, cgroup memory limit, or JVM heap size. However, these values only override the resource capacity advertised to the CoordinatorServer. CPU is currently used for resource reporting, while memory is also used to estimate the cluster-level KV leader replica capacity. Could we rename them to make the advertised-only semantics explicit? - `tablet-server.advertised-resource.cpu-cores` - `tablet-server.advertised-resource.memory-size` The Java constants could be renamed accordingly: - `TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES` - `TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE` The descriptions should also explicitly state that these options do not enforce resource limits. For example: ```java "The CPU capacity, in cores, that this tablet server advertises " + "to the CoordinatorServer for resource reporting. " + "This option does not limit CPU usage or configure a cgroup CPU quota. " + "If not configured, the tablet server detects the value from " + "cgroup CPU quota or the JVM runtime." ``` ```java "The memory capacity that this tablet server advertises " + "to the CoordinatorServer for resource reporting and " + "cluster-level KV leader replica capacity estimation. " + "This option does not configure JVM heap size, reserve memory, " + "or enforce a process or container memory limit. " + "It represents total usable capacity, not current memory usage " + "or free memory. If not configured, the tablet server detects " + "the value from cgroup or operating system information." ``` Using `advertised-resource` makes it clear that these values are informational inputs for reporting and capacity planning, rather than resource limits enforced by the TabletServer. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/KvLeaderReplicaCapacityManager.java: ########## @@ -0,0 +1,221 @@ +/* + * 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.fluss.server.coordinator; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.cluster.ServerReconfigurable; +import org.apache.fluss.exception.ConfigException; +import org.apache.fluss.exception.DatabaseNotExistException; +import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.server.metadata.CoordinatorMetadataCache; +import org.apache.fluss.server.metadata.ServerInfo; + +import java.util.List; +import java.util.Set; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Manages the in-memory cluster capacity for KV leader replicas. */ +public class KvLeaderReplicaCapacityManager implements ServerReconfigurable { + + /** Negative capacity means the automatic capacity limit is disabled. */ + public static final long CAPACITY_LIMIT_DISABLED = -1L; + + private final Object lock = new Object(); + private final CoordinatorMetadataCache metadataCache; + + private long leaderReplicaMemoryReservedBytes; + private long currentKvLeaderReplicaCount; + + public KvLeaderReplicaCapacityManager( + Configuration conf, CoordinatorMetadataCache metadataCache) { + this.metadataCache = checkNotNull(metadataCache, "metadataCache should not be null."); + this.leaderReplicaMemoryReservedBytes = getLeaderReplicaMemoryReservedBytes(conf); + } + + @Override + public void validate(Configuration newConfig) throws ConfigException { + long memoryReservedBytes = getLeaderReplicaMemoryReservedBytes(newConfig); + if (memoryReservedBytes <= 0) { + throw new ConfigException( + ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED.key() + + " must be greater than 0."); + } Review Comment: Is it possible (how to) to disable the leader-replica memory reserve? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
