You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/07/17 02:40:03 UTC
(fluss) branch main updated: [server] Add resource-based KV leader replica capacity control (#3614)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 2c8b9753a [server] Add resource-based KV leader replica capacity control (#3614)
2c8b9753a is described below
commit 2c8b9753a6823b4b55682ebab0c141e63da44afb
Author: yunhong <33...@qq.com>
AuthorDate: Fri Jul 17 10:39:59 2026 +0800
[server] Add resource-based KV leader replica capacity control (#3614)
---
.../org/apache/fluss/config/ConfigOptions.java | 34 ++++
...sufficientKvLeaderReplicaCapacityException.java | 36 ++++
.../java/org/apache/fluss/metrics/MetricNames.java | 2 +
.../java/org/apache/fluss/rpc/protocol/Errors.java | 7 +-
.../apache/fluss/server/DynamicServerConfig.java | 2 +
.../server/coordinator/AutoPartitionManager.java | 19 +-
.../server/coordinator/CoordinatorContext.java | 17 ++
.../coordinator/CoordinatorEventProcessor.java | 59 ++++--
.../server/coordinator/CoordinatorServer.java | 23 ++-
.../server/coordinator/CoordinatorService.java | 52 ++++-
.../fluss/server/coordinator/MetadataManager.java | 2 +-
.../coordinator/ReplicaCapacityController.java | 184 +++++++++++++++++
.../fluss/server/coordinator/TableManager.java | 23 ++-
.../event/watcher/TabletServerChangeWatcher.java | 3 +-
.../server/metadata/CoordinatorMetadataCache.java | 14 ++
.../apache/fluss/server/metadata/ServerInfo.java | 22 +-
.../server/metadata/TabletServerResource.java | 80 ++++++++
.../apache/fluss/server/tablet/TabletServer.java | 7 +-
.../server/tablet/TabletServerResourceProbe.java | 199 ++++++++++++++++++
.../server/zk/data/TabletServerRegistration.java | 22 +-
.../zk/data/TabletServerRegistrationJsonSerde.java | 20 +-
.../fluss/server/DynamicConfigChangeTest.java | 58 ++++++
.../coordinator/AutoPartitionManagerTest.java | 135 +++++++++++++
.../server/coordinator/CoordinatorContextTest.java | 20 ++
.../coordinator/CoordinatorEventProcessorTest.java | 198 +++++++++++++++++-
.../coordinator/ReplicaCapacityControllerTest.java | 224 +++++++++++++++++++++
.../server/coordinator/TableManagerITCase.java | 169 ++++++++++++++++
.../fluss/server/coordinator/TableManagerTest.java | 12 ++
.../rebalance/RebalanceManagerTest.java | 9 +-
.../statemachine/TableBucketStateMachineTest.java | 9 +-
.../tablet/TabletServerResourceProbeTest.java | 136 +++++++++++++
.../TabletServerRegistrationJsonSerdeTest.java | 32 ++-
32 files changed, 1779 insertions(+), 50 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index cc7ce32ad..9323ac0a4 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -319,6 +319,18 @@ public class ConfigOptions {
+ "This default is capped to reduce the risk that an assignment znode exceeds "
+ "ZooKeeper's packet size limit.");
+ public static final ConfigOption<MemorySize> KV_LEADER_REPLICA_MEMORY_RESERVED =
+ key("kv.leader-replica.memory-reserved")
+ .memoryType()
+ .defaultValue(MemorySize.ZERO)
+ .withDescription(
+ "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. The default value of 0 "
+ + "disables memory-based KV leader replica capacity control. "
+ + "Set a positive value to enable it.");
+
/**
* The network address and port the server binds to for accepting connections.
*
@@ -590,6 +602,28 @@ 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_ADVERTISED_RESOURCE_CPU_CORES =
+ key("tablet-server.advertised-resource.cpu-cores")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription(
+ "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.");
+
+ public static final ConfigOption<MemorySize> TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE =
+ key("tablet-server.advertised-resource.memory-size")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ "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.");
+
public static final ConfigOption<String> DATA_DIR =
key("data.dir")
.stringType()
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/InsufficientKvLeaderReplicaCapacityException.java b/fluss-common/src/main/java/org/apache/fluss/exception/InsufficientKvLeaderReplicaCapacityException.java
new file mode 100644
index 000000000..1b5a5e6d0
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/InsufficientKvLeaderReplicaCapacityException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.exception;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * This exception is thrown if a KV table or partition creation exceeds the cluster KV leader
+ * replica capacity.
+ *
+ * @since 1.0
+ */
+@PublicEvolving
+public class InsufficientKvLeaderReplicaCapacityException extends ApiException {
+
+ private static final long serialVersionUID = 1L;
+
+ public InsufficientKvLeaderReplicaCapacityException(String message) {
+ super(message);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index dc93a5e3d..3d44f6290 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -43,6 +43,8 @@ public class MetricNames {
public static final String LAKE_TABLE_COUNT = "lakeTableCount";
public static final String BUCKET_COUNT = "bucketCount";
public static final String PARTITION_COUNT = "partitionCount";
+ public static final String KV_LEADER_REPLICA_COUNT = "kvLeaderReplicaCount";
+ public static final String KV_LEADER_REPLICA_CAPACITY = "kvLeaderReplicaCapacity";
public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount";
public static final String PENDING_LEADER_ACTIVATION_COUNT = "pendingLeaderActivationCount";
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
index 0307932ec..9a46b0460 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
@@ -32,6 +32,7 @@ import org.apache.fluss.exception.DuplicateSequenceException;
import org.apache.fluss.exception.FencedLeaderEpochException;
import org.apache.fluss.exception.FencedTieringEpochException;
import org.apache.fluss.exception.IneligibleReplicaException;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidColumnProjectionException;
import org.apache.fluss.exception.InvalidConfigException;
@@ -270,7 +271,11 @@ public enum Errors {
DISK_WRITE_LOCKED(
70,
"The tablet server has rejected writes because its data disk usage reached the configured write-limit ratio.",
- DiskWriteLockedException::new);
+ DiskWriteLockedException::new),
+ INSUFFICIENT_KV_LEADER_REPLICA_CAPACITY(
+ 71,
+ "The cluster does not have enough KV leader replica capacity.",
+ InsufficientKvLeaderReplicaCapacityException::new);
private static final Logger LOG = LoggerFactory.getLogger(Errors.class);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
index 269e0c483..98e845efc 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
@@ -46,6 +46,7 @@ import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
+import static org.apache.fluss.config.ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED;
import static org.apache.fluss.config.ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC;
import static org.apache.fluss.config.ConfigOptions.KV_SNAPSHOT_INTERVAL;
import static org.apache.fluss.config.ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER;
@@ -74,6 +75,7 @@ class DynamicServerConfig {
Arrays.asList(
DATALAKE_FORMAT.key(),
LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
+ KV_LEADER_REPLICA_MEMORY_RESERVED.key(),
KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
KV_SNAPSHOT_INTERVAL.key(),
SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
index d47cf7967..7f513a476 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
@@ -22,6 +22,7 @@ import org.apache.fluss.cluster.TabletServerInfo;
import org.apache.fluss.config.AutoPartitionTimeUnit;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
import org.apache.fluss.exception.PartitionAlreadyExistsException;
import org.apache.fluss.exception.PartitionNotExistException;
import org.apache.fluss.exception.TooManyBucketsException;
@@ -91,6 +92,7 @@ public class AutoPartitionManager implements AutoCloseable {
private final ServerMetadataCache metadataCache;
private final MetadataManager metadataManager;
private final RemoteDirDynamicLoader remoteDirDynamicLoader;
+ private final ReplicaCapacityController replicaCapacityController;
private final Clock clock;
private final long periodicInterval;
@@ -115,12 +117,14 @@ public class AutoPartitionManager implements AutoCloseable {
ServerMetadataCache metadataCache,
MetadataManager metadataManager,
RemoteDirDynamicLoader remoteDirDynamicLoader,
- Configuration conf) {
+ Configuration conf,
+ ReplicaCapacityController replicaCapacityController) {
this(
metadataCache,
metadataManager,
remoteDirDynamicLoader,
conf,
+ replicaCapacityController,
SystemClock.getInstance(),
// TODO: Reuse the CoordinatorServer shared scheduler for this lightweight
// coordinator periodic task instead of creating a component-owned scheduler.
@@ -134,11 +138,13 @@ public class AutoPartitionManager implements AutoCloseable {
MetadataManager metadataManager,
RemoteDirDynamicLoader remoteDirDynamicLoader,
Configuration conf,
+ ReplicaCapacityController replicaCapacityController,
Clock clock,
ScheduledExecutorService periodicExecutor) {
this.metadataCache = metadataCache;
this.metadataManager = metadataManager;
this.remoteDirDynamicLoader = remoteDirDynamicLoader;
+ this.replicaCapacityController = replicaCapacityController;
this.clock = clock;
this.periodicExecutor = periodicExecutor;
this.periodicInterval = conf.get(ConfigOptions.AUTO_PARTITION_CHECK_INTERVAL).toMillis();
@@ -412,7 +418,11 @@ public class AutoPartitionManager implements AutoCloseable {
long tableId = tableInfo.getTableId();
int replicaFactor = tableInfo.getTableConfig().getReplicationFactor();
TabletServerInfo[] servers = metadataCache.getLiveServers();
+ long newKvLeaderReplicaCount =
+ tableInfo.hasPrimaryKey() ? tableInfo.getNumBuckets() : 0;
try {
+ replicaCapacityController.checkCanCreateKvLeaderReplicas(newKvLeaderReplicaCount);
+
Map<Integer, BucketAssignment> bucketAssignments =
generateAssignment(tableInfo.getNumBuckets(), replicaFactor, servers)
.getBucketAssignments();
@@ -446,6 +456,13 @@ public class AutoPartitionManager implements AutoCloseable {
+ "because exceed the maximum number of buckets per partition.",
partition,
tablePath);
+ } catch (InsufficientKvLeaderReplicaCapacityException t) {
+ LOG.warn(
+ "Auto partitioning skip to create partition {} for table [{}], "
+ + "because {}",
+ partition,
+ tablePath,
+ t.getMessage());
} catch (Exception e) {
LOG.error(
"Auto partitioning failed to create partition {} for table [{}].",
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
index 6a5710ad3..1dc6266e5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
@@ -94,6 +94,7 @@ public class CoordinatorContext {
private final Map<Long, TableInfo> tableInfoById = new HashMap<>();
private final Map<TableBucket, LeaderAndIsr> bucketLeaderAndIsr = new HashMap<>();
+ private final Set<TableBucket> kvBuckets = new HashSet<>();
private final Set<Long> tablesToBeDeleted = new HashSet<>();
private final Set<TablePartition> partitionsToBeDeleted = new HashSet<>();
@@ -347,6 +348,21 @@ public class CoordinatorContext {
return allBuckets;
}
+ /** Idempotently adds buckets observed as KV buckets. */
+ public void addKvBuckets(Collection<TableBucket> tableBuckets) {
+ kvBuckets.addAll(tableBuckets);
+ }
+
+ /** Idempotently removes buckets that are no longer observed as KV buckets. */
+ public void removeKvBuckets(Collection<TableBucket> tableBuckets) {
+ kvBuckets.removeAll(tableBuckets);
+ }
+
+ /** Returns the number of KV buckets currently observed by the Coordinator. */
+ public long getKvBucketCount() {
+ return kvBuckets.size();
+ }
+
public Set<TableBucketReplica> replicasOnTabletServer(int server) {
Set<TableBucketReplica> replicasInServer = new HashSet<>();
tableAssignments.forEach(
@@ -818,6 +834,7 @@ public class CoordinatorContext {
tableInfoById.clear();
pathByPartitionId.clear();
partitionIdByPath.clear();
+ kvBuckets.clear();
}
public void resetContext() {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 18c4429af..bac833c3b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -174,6 +174,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
private final ZooKeeperClient zooKeeperClient;
private final ExecutorService ioExecutor;
private final CoordinatorContext coordinatorContext;
+ private final ReplicaCapacityController replicaCapacityController;
private final ReplicaStateMachine replicaStateMachine;
private final TableBucketStateMachine tableBucketStateMachine;
private final CoordinatorEventManager coordinatorEventManager;
@@ -202,6 +203,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
CoordinatorMetadataCache serverMetadataCache,
CoordinatorChannelManager coordinatorChannelManager,
CoordinatorContext coordinatorContext,
+ ReplicaCapacityController replicaCapacityController,
AutoPartitionManager autoPartitionManager,
LakeTableTieringManager lakeTableTieringManager,
CoordinatorMetricGroup coordinatorMetricGroup,
@@ -215,6 +217,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
this.serverMetadataCache = serverMetadataCache;
this.coordinatorChannelManager = coordinatorChannelManager;
this.coordinatorContext = coordinatorContext;
+ this.replicaCapacityController = replicaCapacityController;
this.coordinatorEventManager = new CoordinatorEventManager(this, coordinatorMetricGroup);
this.replicaStateMachine =
new ReplicaStateMachine(
@@ -239,6 +242,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
new TableManager(
metadataManager,
coordinatorContext,
+ replicaCapacityController,
replicaStateMachine,
tableBucketStateMachine,
new RemoteStorageCleaner(conf, ioExecutor),
@@ -319,20 +323,6 @@ public class CoordinatorEventProcessor implements EventProcessor {
throw new FlussRuntimeException("Fail to initialize coordinator context.", e);
}
- // We need to send UpdateMetadataRequest after the coordinator context is initialized and
- // before the state machines in tableManager are started. This is because tablet servers
- // need to receive the list of live tablet servers from UpdateMetadataRequest before they
- // can process the LeaderRequests that are generated by replicaStateMachine.startup() and
- // partitionStateMachine.startup().
- // update coordinator metadata cache when CoordinatorServer start.
- HashSet<ServerInfo> tabletServerInfoList =
- new HashSet<>(coordinatorContext.getLiveTabletServers().values());
- serverMetadataCache.updateMetadata(
- coordinatorContext.getCoordinatorServerInfo(),
- tabletServerInfoList,
- coordinatorContext.getServerTags());
- updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList);
-
lifecycleThrottler.start();
// start table manager
@@ -353,6 +343,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
rebalanceManager.close();
onShutdown();
coordinatorContext.resetContext();
+ updateObservedKvLeaderReplicaCount();
}
private ServerInfo getCoordinatorServerInfo() {
@@ -403,7 +394,8 @@ public class CoordinatorEventProcessor implements EventProcessor {
server,
registration.getRack(),
registration.getEndpoints(),
- ServerType.TABLET_SERVER);
+ ServerType.TABLET_SERVER,
+ registration.getResource());
// Get internal listener endpoint to send request to tablet server.
Endpoint internalEndpoint = serverInfo.endpoint(internalListenerName);
if (internalEndpoint == null) {
@@ -485,7 +477,6 @@ public class CoordinatorEventProcessor implements EventProcessor {
"Load tables success in {}ms when initializing coordinator context.",
System.currentTimeMillis() - start4loadTables);
- autoPartitionManager.initAutoPartitionTables(autoPartitionTables);
lakeTableTieringManager.initWithLakeTables(lakeTables);
// load all assignment
@@ -496,6 +487,25 @@ public class CoordinatorEventProcessor implements EventProcessor {
"Load table and partition assignment success in {}ms when initializing coordinator context.",
System.currentTimeMillis() - start4loadAssignment);
+ updateObservedKvLeaderReplicaCount();
+
+ // We need to send UpdateMetadataRequest after the coordinator context is initialized and
+ // before the state machines in tableManager are started. This is because tablet servers
+ // need to receive the list of live tablet servers from UpdateMetadataRequest before they
+ // can process the LeaderRequests that are generated by replicaStateMachine.startup() and
+ // partitionStateMachine.startup().
+ HashSet<ServerInfo> tabletServerInfoList =
+ new HashSet<>(coordinatorContext.getLiveTabletServers().values());
+ serverMetadataCache.updateMetadata(
+ coordinatorContext.getCoordinatorServerInfo(),
+ tabletServerInfoList,
+ coordinatorContext.getServerTags());
+ updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList);
+
+ // Auto-partition initialization schedules creation checks immediately. Start it only after
+ // the observed KV leader replica count and live tablet server resources are restored.
+ autoPartitionManager.initAutoPartitionTables(autoPartitionTables);
+
long end = System.currentTimeMillis();
LOG.info("Current total {} tables in the cluster.", coordinatorContext.allTables().size());
LOG.info(
@@ -572,6 +582,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
coordinatorContext.updateBucketReplicaAssignment(
tableBucket, bucketAssignment.getReplicas());
}
+ trackKvBucketsForLoadedAssignment(tableId, tableBucketSet);
Map<TableBucket, LeaderAndIsr> leaderAndIsrMap =
zooKeeperClient.getLeaderAndIsrs(tableBucketSet);
for (TableBucket tableBucket : tableBucketSet) {
@@ -597,6 +608,17 @@ public class CoordinatorEventProcessor implements EventProcessor {
}
}
+ @VisibleForTesting
+ void trackKvBucketsForLoadedAssignment(long tableId, Set<TableBucket> tableBuckets) {
+ TableInfo tableInfo = coordinatorContext.getTableInfoById(tableId);
+ // During failover, an orphan assignment may outlive its table metadata. Conservatively
+ // treat it as KV until lifecycle cleanup removes the buckets to avoid underestimating
+ // the KV leader replica count.
+ if (tableInfo == null || tableInfo.hasPrimaryKey()) {
+ coordinatorContext.addKvBuckets(tableBuckets);
+ }
+ }
+
private void onShutdown() {
// first shutdown table manager
tableManager.shutdown();
@@ -760,6 +782,11 @@ public class CoordinatorEventProcessor implements EventProcessor {
}
}
+ private void updateObservedKvLeaderReplicaCount() {
+ replicaCapacityController.updateObservedKvLeaderReplicaCount(
+ coordinatorContext.getKvBucketCount());
+ }
+
private void processRetryOfflineLeader() {
// This event consumes the currently scheduled one-shot retry. If the NotifyLeaderAndIsr
// request sent below is rejected by the target tablet server again, its response handler
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
index 3cbaac40d..bd66a1a9d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -165,6 +165,9 @@ public class CoordinatorServer extends ServerBase {
@GuardedBy("lock")
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
+ @GuardedBy("lock")
+ private ReplicaCapacityController replicaCapacityController;
+
public CoordinatorServer(Configuration conf) {
this(conf, SystemClock.getInstance());
}
@@ -240,9 +243,11 @@ public class CoordinatorServer extends ServerBase {
this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true);
this.remoteDirDynamicLoader = new RemoteDirDynamicLoader(conf);
+ this.metadataCache = new CoordinatorMetadataCache();
+ this.replicaCapacityController =
+ new ReplicaCapacityController(conf, metadataCache, serverMetricGroup);
this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf);
- this.metadataCache = new CoordinatorMetadataCache();
this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);
if (authorizer != null) {
@@ -285,7 +290,8 @@ public class CoordinatorServer extends ServerBase {
dynamicConfigManager,
ioExecutor,
kvSnapshotLeaseManager,
- coordinatorLeaderElection);
+ coordinatorLeaderElection,
+ replicaCapacityController);
this.rpcServer =
RpcServer.create(
@@ -298,6 +304,7 @@ public class CoordinatorServer extends ServerBase {
// Register server reconfigurable components
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.register(remoteDirDynamicLoader);
+ dynamicConfigManager.register(replicaCapacityController);
// Register stateless validators for coordinator-side upfront validation
dynamicConfigManager.register(new DiskWriteLimitConfigValidator());
rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register);
@@ -324,7 +331,11 @@ public class CoordinatorServer extends ServerBase {
this.autoPartitionManager =
new AutoPartitionManager(
- metadataCache, metadataManager, remoteDirDynamicLoader, conf);
+ metadataCache,
+ metadataManager,
+ remoteDirDynamicLoader,
+ conf,
+ replicaCapacityController);
autoPartitionManager.start();
// start coordinator event processor after we register coordinator leader to zk
@@ -338,6 +349,7 @@ public class CoordinatorServer extends ServerBase {
metadataCache,
coordinatorChannelManager,
coordinatorContext,
+ replicaCapacityController,
autoPartitionManager,
lakeTableTieringManager,
serverMetricGroup,
@@ -556,6 +568,11 @@ public class CoordinatorServer extends ServerBase {
}
}
+ @VisibleForTesting
+ ReplicaCapacityController getReplicaCapacityController() {
+ return replicaCapacityController;
+ }
+
CompletableFuture<Void> stopServices() {
synchronized (lock) {
Throwable exception = null;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index 0772f5e3d..ef6a73567 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -257,6 +257,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
private final KvSnapshotLeaseManager kvSnapshotLeaseManager;
private final CoordinatorLeaderElection coordinatorLeaderElection;
private final RemoteDirDynamicLoader remoteDirDynamicLoader;
+ private final ReplicaCapacityController replicaCapacityController;
public CoordinatorService(
Configuration conf,
@@ -272,7 +273,8 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
DynamicConfigManager dynamicConfigManager,
ExecutorService ioExecutor,
KvSnapshotLeaseManager kvSnapshotLeaseManager,
- CoordinatorLeaderElection coordinatorLeaderElection) {
+ CoordinatorLeaderElection coordinatorLeaderElection,
+ ReplicaCapacityController replicaCapacityController) {
super(
remoteFileSystem,
ServerType.COORDINATOR,
@@ -305,6 +307,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
this.kvSnapshotLeaseManager = kvSnapshotLeaseManager;
this.coordinatorLeaderElection = coordinatorLeaderElection;
+ this.replicaCapacityController = replicaCapacityController;
}
@Override
@@ -485,6 +488,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
// the distribution and bucket count must be set now
//noinspection OptionalGetWithoutIsPresent
int bucketCount = tableDescriptor.getTableDistribution().get().getBucketCount().get();
+ long newKvLeaderReplicaCount = getNewKvLeaderReplicaCount(tableDescriptor);
// first, generate the assignment
TableAssignment tableAssignment = null;
@@ -496,6 +500,12 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
tableAssignment = generateAssignment(bucketCount, replicaFactor, servers);
}
+ if (request.isIgnoreIfExists() && metadataManager.tableExists(tablePath)) {
+ return CompletableFuture.completedFuture(new CreateTableResponse());
+ }
+
+ replicaCapacityController.checkCanCreateKvLeaderReplicas(newKvLeaderReplicaCount);
+
// before create table in fluss, we may create in lake
if (isDataLakeEnabled(tableDescriptor)) {
try {
@@ -529,6 +539,15 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
return CompletableFuture.completedFuture(new CreateTableResponse());
}
+ private long getNewKvLeaderReplicaCount(TableDescriptor tableDescriptor) {
+ if (!tableDescriptor.hasPrimaryKey() || tableDescriptor.isPartitioned()) {
+ return 0;
+ }
+ // the distribution and bucket count must be set now
+ //noinspection OptionalGetWithoutIsPresent
+ return tableDescriptor.getTableDistribution().get().getBucketCount().get();
+ }
+
@Override
public CompletableFuture<AlterTableResponse> alterTable(AlterTableRequest request) {
TablePath tablePath = toTablePath(request.getTablePath());
@@ -715,40 +734,51 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
authorizeTable(OperationType.WRITE, tablePath);
CreatePartitionResponse response = new CreatePartitionResponse();
- TableRegistration table = metadataManager.getTableRegistration(tablePath);
- if (!table.isPartitioned()) {
+ TableInfo tableInfo = metadataManager.getTable(tablePath);
+ if (!tableInfo.isPartitioned()) {
throw new TableNotPartitionedException(
"Only partitioned table support create partition.");
}
// first, validate the partition spec, and get resolved partition spec.
PartitionSpec partitionSpec = getPartitionSpec(request.getPartitionSpec());
- validatePartitionSpec(tablePath, table.partitionKeys, partitionSpec, true);
+ validatePartitionSpec(tablePath, tableInfo.getPartitionKeys(), partitionSpec, true);
// second, check whether the partition is out-of-date.
validateAutoPartitionTime(
partitionSpec,
- table.partitionKeys,
- table.getTableConfig().getAutoPartitionStrategy());
+ tableInfo.getPartitionKeys(),
+ tableInfo.getTableConfig().getAutoPartitionStrategy());
ResolvedPartitionSpec partitionToCreate =
- ResolvedPartitionSpec.fromPartitionSpec(table.partitionKeys, partitionSpec);
+ ResolvedPartitionSpec.fromPartitionSpec(
+ tableInfo.getPartitionKeys(), partitionSpec);
+ if (request.isIgnoreIfNotExists()
+ && metadataManager
+ .getOptionalPartitionRegistration(
+ tablePath, partitionToCreate.getPartitionName())
+ .isPresent()) {
+ return CompletableFuture.completedFuture(response);
+ }
+
+ long newKvLeaderReplicaCount = tableInfo.hasPrimaryKey() ? tableInfo.getNumBuckets() : 0;
+ replicaCapacityController.checkCanCreateKvLeaderReplicas(newKvLeaderReplicaCount);
// third, generate the PartitionAssignment.
- int replicaFactor = table.getTableConfig().getReplicationFactor();
+ int replicaFactor = tableInfo.getTableConfig().getReplicationFactor();
TabletServerInfo[] servers = metadataCache.getLiveServers();
Map<Integer, BucketAssignment> bucketAssignments =
- generateAssignment(table.bucketCount, replicaFactor, servers)
+ generateAssignment(tableInfo.getNumBuckets(), replicaFactor, servers)
.getBucketAssignments();
PartitionAssignment partitionAssignment =
- new PartitionAssignment(table.tableId, bucketAssignments);
+ new PartitionAssignment(tableInfo.getTableId(), bucketAssignments);
// select remote data dir for partition
String remoteDataDir = remoteDirDynamicLoader.getRemoteDirSelector().nextDataDir();
metadataManager.createPartition(
tablePath,
- table.tableId,
+ tableInfo.getTableId(),
remoteDataDir,
partitionAssignment,
partitionToCreate,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index 318424882..48533ea0d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -910,7 +910,7 @@ public class MetadataManager {
}
}
- private Optional<PartitionRegistration> getOptionalPartitionRegistration(
+ Optional<PartitionRegistration> getOptionalPartitionRegistration(
TablePath tablePath, String partitionName) {
try {
return zookeeperClient.getPartition(tablePath, partitionName);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCapacityController.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCapacityController.java
new file mode 100644
index 000000000..3ffa15ba1
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ReplicaCapacityController.java
@@ -0,0 +1,184 @@
+/*
+ * 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.InsufficientKvLeaderReplicaCapacityException;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Controls replica creation admission based on in-memory cluster capacity. */
+public class ReplicaCapacityController implements ServerReconfigurable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReplicaCapacityController.class);
+
+ /** Negative capacity means the automatic capacity limit is disabled. */
+ public static final long CAPACITY_LIMIT_DISABLED = -1L;
+
+ private final CoordinatorMetadataCache metadataCache;
+
+ private volatile long kvLeaderReplicaMemoryReservedBytes;
+ private volatile long observedKvLeaderReplicaCount;
+
+ public ReplicaCapacityController(
+ Configuration conf,
+ CoordinatorMetadataCache metadataCache,
+ CoordinatorMetricGroup coordinatorMetricGroup) {
+ this(conf, metadataCache);
+ registerMetrics(
+ checkNotNull(coordinatorMetricGroup, "coordinatorMetricGroup should not be null."));
+ }
+
+ @VisibleForTesting
+ ReplicaCapacityController(Configuration conf, CoordinatorMetadataCache metadataCache) {
+ this.metadataCache = checkNotNull(metadataCache, "metadataCache should not be null.");
+ long memoryReservedBytes = getKvLeaderReplicaMemoryReservedBytes(conf);
+ validateKvLeaderReplicaMemoryReservedBytes(memoryReservedBytes);
+ this.kvLeaderReplicaMemoryReservedBytes = memoryReservedBytes;
+ }
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ long memoryReservedBytes = getKvLeaderReplicaMemoryReservedBytes(newConfig);
+ validateKvLeaderReplicaMemoryReservedBytes(memoryReservedBytes);
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) throws ConfigException {
+ kvLeaderReplicaMemoryReservedBytes = getKvLeaderReplicaMemoryReservedBytes(newConfig);
+ }
+
+ /** Checks whether the requested KV leader replicas fit the current observed capacity. */
+ public void checkCanCreateKvLeaderReplicas(long requestedKvLeaderReplicaCount) {
+ if (requestedKvLeaderReplicaCount <= 0) {
+ return;
+ }
+
+ long capacity = getKvLeaderReplicaCapacity();
+ long observedCount = observedKvLeaderReplicaCount;
+ if (capacity != CAPACITY_LIMIT_DISABLED
+ && (observedCount > capacity
+ || requestedKvLeaderReplicaCount > capacity - observedCount)) {
+ throw new InsufficientKvLeaderReplicaCapacityException(
+ String.format(
+ "Not enough KV leader replica capacity. "
+ + "observedKvLeaderReplicaCount=%s, "
+ + "requestedKvLeaderReplicaCount=%s, "
+ + "kvLeaderReplicaCapacity=%s.",
+ observedCount, requestedKvLeaderReplicaCount, capacity));
+ }
+ }
+
+ /** Replaces the observed KV leader replica count with a CoordinatorContext-derived snapshot. */
+ public void updateObservedKvLeaderReplicaCount(long observedKvLeaderReplicaCount) {
+ checkArgument(
+ observedKvLeaderReplicaCount >= 0,
+ "Observed KV leader replica count must be greater than or equal to 0.");
+ this.observedKvLeaderReplicaCount = observedKvLeaderReplicaCount;
+ }
+
+ /** Returns the latest CoordinatorContext-derived KV leader replica count. */
+ public long getKvLeaderReplicaCount() {
+ return observedKvLeaderReplicaCount;
+ }
+
+ /** Returns the current cluster KV leader replica capacity, or -1 if disabled. */
+ public long getKvLeaderReplicaCapacity() {
+ long memoryReservedBytes = kvLeaderReplicaMemoryReservedBytes;
+ if (memoryReservedBytes == 0) {
+ return CAPACITY_LIMIT_DISABLED;
+ }
+
+ 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);
+ long capacity = totalMemoryBytes / memoryReservedBytes;
+ LOG.debug(
+ "Calculated KV leader replica capacity: liveTabletServerCount={}, "
+ + "knownMemoryTabletServerCount={}, knownMemoryBytes={}, "
+ + "averageKnownMemoryBytes={}, estimatedTotalMemoryBytes={}, "
+ + "kvLeaderReplicaMemoryReservedBytes={}, kvLeaderReplicaCapacity={}.",
+ liveTabletServerCount,
+ knownMemoryTabletServerCount,
+ knownMemoryBytes,
+ averageKnownMemoryBytes,
+ totalMemoryBytes,
+ memoryReservedBytes,
+ capacity);
+ return capacity;
+ }
+
+ @VisibleForTesting
+ long getKvLeaderReplicaMemoryReservedBytes() {
+ return kvLeaderReplicaMemoryReservedBytes;
+ }
+
+ private void registerMetrics(CoordinatorMetricGroup coordinatorMetricGroup) {
+ coordinatorMetricGroup.gauge(
+ MetricNames.KV_LEADER_REPLICA_COUNT, this::getKvLeaderReplicaCount);
+ coordinatorMetricGroup.gauge(
+ MetricNames.KV_LEADER_REPLICA_CAPACITY, this::getKvLeaderReplicaCapacity);
+ }
+
+ private static long getKvLeaderReplicaMemoryReservedBytes(Configuration conf) {
+ return conf.get(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED).getBytes();
+ }
+
+ private static void validateKvLeaderReplicaMemoryReservedBytes(long memoryReservedBytes) {
+ if (memoryReservedBytes < 0) {
+ throw new ConfigException(
+ ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED.key()
+ + " must be greater than or equal to 0.");
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
index 978751ec4..2381447bc 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableManager.java
@@ -47,6 +47,7 @@ public class TableManager {
private final MetadataManager metadataManager;
private final RemoteStorageCleaner remoteStorageCleaner;
private final CoordinatorContext coordinatorContext;
+ private final ReplicaCapacityController replicaCapacityController;
private final ReplicaStateMachine replicaStateMachine;
private final TableBucketStateMachine tableBucketStateMachine;
private final ExecutorService ioExecutor;
@@ -66,6 +67,7 @@ public class TableManager {
public TableManager(
MetadataManager metadataManager,
CoordinatorContext coordinatorContext,
+ ReplicaCapacityController replicaCapacityController,
ReplicaStateMachine replicaStateMachine,
TableBucketStateMachine tableBucketStateMachine,
RemoteStorageCleaner remoteStorageCleaner,
@@ -74,6 +76,7 @@ public class TableManager {
this.metadataManager = metadataManager;
this.remoteStorageCleaner = remoteStorageCleaner;
this.coordinatorContext = coordinatorContext;
+ this.replicaCapacityController = replicaCapacityController;
this.replicaStateMachine = replicaStateMachine;
this.tableBucketStateMachine = tableBucketStateMachine;
this.ioExecutor = ioExecutor;
@@ -165,6 +168,11 @@ public class TableManager {
"New table buckets: {} for table {}.",
tableBuckets,
coordinatorContext.getTablePathById(tableId));
+ TableInfo tableInfo = coordinatorContext.getTableInfoById(tableId);
+ if (tableInfo != null && tableInfo.hasPrimaryKey()) {
+ coordinatorContext.addKvBuckets(tableBuckets);
+ updateObservedKvLeaderReplicaCount();
+ }
// first, we transmit it to state NewBucket
tableBucketStateMachine.handleStateChange(tableBuckets, BucketState.NewBucket);
// then get all the replicas of the all table buckets
@@ -182,7 +190,7 @@ public class TableManager {
Set<TableBucket> tableBuckets = coordinatorContext.getAllBucketsForTable(tableId);
tableBucketStateMachine.handleStateChange(tableBuckets, BucketState.OfflineBucket);
tableBucketStateMachine.handleStateChange(tableBuckets, BucketState.NonExistentBucket);
- onDeleteTableBucket(coordinatorContext.getAllReplicasForTable(tableId));
+ onDeleteTableBucket(tableBuckets, coordinatorContext.getAllReplicasForTable(tableId));
}
/** Invoked with partitions of a table to be deleted. */
@@ -191,7 +199,8 @@ public class TableManager {
coordinatorContext.getAllBucketsForPartition(tableId, partitionId);
tableBucketStateMachine.handleStateChange(deleteBuckets, BucketState.OfflineBucket);
tableBucketStateMachine.handleStateChange(deleteBuckets, BucketState.NonExistentBucket);
- onDeleteTableBucket(coordinatorContext.getAllReplicasForPartition(tableId, partitionId));
+ onDeleteTableBucket(
+ deleteBuckets, coordinatorContext.getAllReplicasForPartition(tableId, partitionId));
}
/**
@@ -207,7 +216,10 @@ public class TableManager {
* with delete=true which will delete all persistent data from all the replicas of the all the
* respective buckets.
*/
- private void onDeleteTableBucket(Set<TableBucketReplica> allReplicas) {
+ private void onDeleteTableBucket(
+ Set<TableBucket> tableBuckets, Set<TableBucketReplica> allReplicas) {
+ coordinatorContext.removeKvBuckets(tableBuckets);
+ updateObservedKvLeaderReplicaCount();
// to offline, send stop replica to all followers that are not in the OfflineReplica state
// so they stop sending fetch requests to the leader
replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.OfflineReplica);
@@ -215,6 +227,11 @@ public class TableManager {
replicaStateMachine.handleStateChanges(allReplicas, ReplicaState.ReplicaDeletionStarted);
}
+ private void updateObservedKvLeaderReplicaCount() {
+ replicaCapacityController.updateObservedKvLeaderReplicaCount(
+ coordinatorContext.getKvBucketCount());
+ }
+
public void resumeDeletions() {
resumeTableDeletions();
resumePartitionDeletions();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TabletServerChangeWatcher.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TabletServerChangeWatcher.java
index 569d5117a..6fe5e39e5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TabletServerChangeWatcher.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/watcher/TabletServerChangeWatcher.java
@@ -86,7 +86,8 @@ public class TabletServerChangeWatcher extends ServerBaseChangeWatcher {
serverId,
registration.getRack(),
registration.getEndpoints(),
- ServerType.TABLET_SERVER)));
+ ServerType.TABLET_SERVER,
+ registration.getResource())));
}
break;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataCache.java
index 013b8759f..74e9f6675 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataCache.java
@@ -94,6 +94,20 @@ public class CoordinatorMetadataCache implements ServerMetadataCache {
return Collections.unmodifiableSet(tabletServerInfos);
}
+ public Set<ServerInfo> getLiveTabletServerInfos() {
+ NodeMetadataSnapshot snapshot = metadataSnapshot;
+ Map<Integer, ServerInfo> aliveTabletServers = snapshot.aliveTabletServers;
+ Set<ServerInfo> serverInfos = new HashSet<>();
+ aliveTabletServers.forEach(
+ (serverId, serverInfo) -> {
+ ServerTag tag = snapshot.serverTags.get(serverId);
+ if (tag != ServerTag.PERMANENT_OFFLINE) {
+ serverInfos.add(serverInfo);
+ }
+ });
+ return Collections.unmodifiableSet(serverInfos);
+ }
+
/**
* Servers with {@code PERMANENT_OFFLINE} tags are no longer returned here. So that no new
* replicas will be assigned to these servers.
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerInfo.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerInfo.java
index bb550bd4e..9e6cdcc26 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerInfo.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ServerInfo.java
@@ -44,15 +44,26 @@ public class ServerInfo {
private final Map<String, Endpoint> endpointMap;
private final ServerType serverType;
+ private final TabletServerResource resource;
public ServerInfo(
Integer id, @Nullable String rack, List<Endpoint> endpoints, ServerType serverType) {
+ this(id, rack, endpoints, serverType, TabletServerResource.unknown());
+ }
+
+ public ServerInfo(
+ Integer id,
+ @Nullable String rack,
+ List<Endpoint> endpoints,
+ ServerType serverType,
+ TabletServerResource resource) {
this.id = id;
this.rack = rack;
this.endpointMap =
endpoints.stream()
.collect(Collectors.toMap(Endpoint::getListenerName, endpoint -> endpoint));
this.serverType = serverType;
+ this.resource = resource;
}
public Integer id() {
@@ -97,6 +108,10 @@ public class ServerInfo {
return new ArrayList<>(endpointMap.values());
}
+ public TabletServerResource resource() {
+ return resource;
+ }
+
@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
@@ -105,12 +120,13 @@ public class ServerInfo {
ServerInfo that = (ServerInfo) o;
return Objects.equals(id, that.id)
&& Objects.equals(rack, that.rack)
- && Objects.equals(endpointMap, that.endpointMap);
+ && Objects.equals(endpointMap, that.endpointMap)
+ && Objects.equals(resource, that.resource);
}
@Override
public int hashCode() {
- return Objects.hash(id, rack, endpointMap);
+ return Objects.hash(id, rack, endpointMap, resource);
}
@Override
@@ -124,6 +140,8 @@ public class ServerInfo {
+ endpointMap.values()
+ ", type="
+ serverType
+ + ", resource="
+ + resource
+ '}';
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerResource.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerResource.java
new file mode 100644
index 000000000..22ce65309
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerResource.java
@@ -0,0 +1,80 @@
+/*
+ * 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.metadata;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/** Resource information reported by a tablet server. */
+public class TabletServerResource {
+
+ private static final TabletServerResource UNKNOWN = new TabletServerResource(null, null);
+
+ private final @Nullable Double cpuCores;
+ private final @Nullable Long memoryBytes;
+
+ public TabletServerResource(@Nullable Double cpuCores, @Nullable Long memoryBytes) {
+ this.cpuCores = cpuCores;
+ this.memoryBytes = memoryBytes;
+ }
+
+ public static TabletServerResource unknown() {
+ return UNKNOWN;
+ }
+
+ public @Nullable Double getCpuCores() {
+ return cpuCores;
+ }
+
+ public @Nullable Long getMemoryBytes() {
+ return memoryBytes;
+ }
+
+ public boolean hasMemory() {
+ return memoryBytes != null && memoryBytes > 0;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TabletServerResource that = (TabletServerResource) o;
+ return Objects.equals(cpuCores, that.cpuCores)
+ && Objects.equals(memoryBytes, that.memoryBytes);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(cpuCores, memoryBytes);
+ }
+
+ @Override
+ public String toString() {
+ return "TabletServerResource{"
+ + "cpuCores="
+ + cpuCores
+ + ", memoryBytes="
+ + memoryBytes
+ + '}';
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 12a18a70f..b38deea91 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -45,6 +45,7 @@ import org.apache.fluss.server.kv.snapshot.DefaultCompletedKvSnapshotCommitter;
import org.apache.fluss.server.log.LogManager;
import org.apache.fluss.server.log.remote.RemoteLogManager;
import org.apache.fluss.server.metadata.TabletServerMetadataCache;
+import org.apache.fluss.server.metadata.TabletServerResource;
import org.apache.fluss.server.metrics.ServerMetricUtils;
import org.apache.fluss.server.metrics.UserMetrics;
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
@@ -380,9 +381,13 @@ public class TabletServer extends ServerBase {
private void registerTabletServer() throws Exception {
long startTime = System.currentTimeMillis();
List<Endpoint> bindEndpoints = rpcServer.getBindEndpoints();
+ TabletServerResource tabletServerResource = new TabletServerResourceProbe(conf).probe();
TabletServerRegistration tabletServerRegistration =
new TabletServerRegistration(
- rack, Endpoint.loadAdvertisedEndpoints(bindEndpoints, conf), startTime);
+ rack,
+ Endpoint.loadAdvertisedEndpoints(bindEndpoints, conf),
+ startTime,
+ tabletServerResource);
while (true) {
try {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java
new file mode 100644
index 000000000..1ea19d419
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServerResourceProbe.java
@@ -0,0 +1,199 @@
+/*
+ * 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.tablet;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.server.metadata.TabletServerResource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Optional;
+
+/** Probes CPU and memory resource information for tablet server registration. */
+public class TabletServerResourceProbe {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TabletServerResourceProbe.class);
+
+ private static final long UNLIMITED_MEMORY_THRESHOLD = Long.MAX_VALUE / 2;
+
+ private final Configuration conf;
+ private final Path cgroupRoot;
+
+ public TabletServerResourceProbe(Configuration conf) {
+ this(conf, Paths.get("/sys/fs/cgroup"));
+ }
+
+ TabletServerResourceProbe(Configuration conf, Path cgroupRoot) {
+ this.conf = conf;
+ this.cgroupRoot = cgroupRoot;
+ }
+
+ public TabletServerResource probe() {
+ return new TabletServerResource(
+ probeCpuCores().orElse(null), probeMemoryBytes().orElse(null));
+ }
+
+ private Optional<Double> probeCpuCores() {
+ Optional<Double> configuredCpuCores =
+ conf.getOptional(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES);
+ if (configuredCpuCores.isPresent()) {
+ Optional<Double> validConfiguredCpuCores = positiveDouble(configuredCpuCores.get());
+ if (validConfiguredCpuCores.isPresent()) {
+ return validConfiguredCpuCores;
+ }
+ }
+
+ Optional<Double> cgroupCpuCores = probeCgroupCpuCores();
+ if (cgroupCpuCores.isPresent()) {
+ return cgroupCpuCores;
+ }
+
+ return positiveDouble(Runtime.getRuntime().availableProcessors());
+ }
+
+ private Optional<Long> probeMemoryBytes() {
+ Optional<MemorySize> configuredMemory =
+ conf.getOptional(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE);
+ if (configuredMemory.isPresent()) {
+ Optional<Long> validConfiguredMemory = positiveLong(configuredMemory.get().getBytes());
+ if (validConfiguredMemory.isPresent()) {
+ return validConfiguredMemory;
+ }
+ }
+
+ Optional<Long> cgroupMemory = probeCgroupMemoryBytes();
+ if (cgroupMemory.isPresent()) {
+ return cgroupMemory;
+ }
+
+ return probeOperatingSystemMemoryBytes();
+ }
+
+ private Optional<Long> probeCgroupMemoryBytes() {
+ Optional<Long> cgroupV2Memory = readMemoryValue(cgroupRoot.resolve("memory.max"));
+ if (cgroupV2Memory.isPresent()) {
+ return cgroupV2Memory;
+ }
+
+ return readMemoryValue(cgroupRoot.resolve("memory").resolve("memory.limit_in_bytes"));
+ }
+
+ private Optional<Long> readMemoryValue(Path path) {
+ Optional<String> value = readFirstLine(path);
+ if (!value.isPresent()) {
+ return Optional.empty();
+ }
+
+ String trimmedValue = value.get().trim();
+ if ("max".equals(trimmedValue)) {
+ return Optional.empty();
+ }
+
+ try {
+ long memoryBytes = Long.parseLong(trimmedValue);
+ if (memoryBytes <= 0 || memoryBytes >= UNLIMITED_MEMORY_THRESHOLD) {
+ return Optional.empty();
+ }
+ return Optional.of(memoryBytes);
+ } catch (NumberFormatException e) {
+ LOG.debug("Failed to parse memory value from {}: {}", path, trimmedValue, e);
+ return Optional.empty();
+ }
+ }
+
+ private Optional<Double> probeCgroupCpuCores() {
+ Optional<String> cpuMax = readFirstLine(cgroupRoot.resolve("cpu.max"));
+ if (cpuMax.isPresent()) {
+ String[] parts = cpuMax.get().trim().split("\\s+");
+ if (parts.length >= 2 && !"max".equals(parts[0])) {
+ Optional<Long> quota = parseLong(parts[0]);
+ Optional<Long> period = parseLong(parts[1]);
+ if (quota.isPresent()
+ && period.isPresent()
+ && quota.get() > 0
+ && period.get() > 0) {
+ return positiveDouble(((double) quota.get()) / period.get());
+ }
+ }
+ }
+
+ Optional<Long> quota = readLong(cgroupRoot.resolve("cpu").resolve("cpu.cfs_quota_us"));
+ Optional<Long> period = readLong(cgroupRoot.resolve("cpu").resolve("cpu.cfs_period_us"));
+ if (quota.isPresent() && period.isPresent() && quota.get() > 0 && period.get() > 0) {
+ return positiveDouble(((double) quota.get()) / period.get());
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Long> probeOperatingSystemMemoryBytes() {
+ java.lang.management.OperatingSystemMXBean operatingSystemMXBean =
+ ManagementFactory.getOperatingSystemMXBean();
+ if (operatingSystemMXBean instanceof com.sun.management.OperatingSystemMXBean) {
+ long memoryBytes =
+ ((com.sun.management.OperatingSystemMXBean) operatingSystemMXBean)
+ .getTotalPhysicalMemorySize();
+ return positiveLong(memoryBytes);
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Long> readLong(Path path) {
+ Optional<String> value = readFirstLine(path);
+ return value.isPresent() ? parseLong(value.get().trim()) : Optional.empty();
+ }
+
+ private Optional<String> readFirstLine(Path path) {
+ if (!Files.isRegularFile(path)) {
+ return Optional.empty();
+ }
+
+ try {
+ byte[] bytes = Files.readAllBytes(path);
+ String content = new String(bytes, StandardCharsets.UTF_8).trim();
+ return content.isEmpty() ? Optional.empty() : Optional.of(content);
+ } catch (IOException e) {
+ LOG.debug("Failed to read resource file {}", path, e);
+ return Optional.empty();
+ }
+ }
+
+ private Optional<Long> parseLong(String value) {
+ try {
+ return Optional.of(Long.parseLong(value));
+ } catch (NumberFormatException e) {
+ return Optional.empty();
+ }
+ }
+
+ private Optional<Long> positiveLong(long value) {
+ return value > 0 ? Optional.of(value) : Optional.empty();
+ }
+
+ private Optional<Double> positiveDouble(double value) {
+ return value > 0 ? Optional.of(value) : Optional.empty();
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistration.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistration.java
index 314a70a8d..5ea57e846 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistration.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistration.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.zk.data;
import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.server.metadata.TabletServerResource;
import javax.annotation.Nullable;
@@ -33,12 +34,22 @@ public class TabletServerRegistration {
private final @Nullable String rack;
private final List<Endpoint> endpoints;
private final long registerTimestamp;
+ private final TabletServerResource resource;
public TabletServerRegistration(
@Nullable String rack, List<Endpoint> endpoints, long registerTimestamp) {
+ this(rack, endpoints, registerTimestamp, TabletServerResource.unknown());
+ }
+
+ public TabletServerRegistration(
+ @Nullable String rack,
+ List<Endpoint> endpoints,
+ long registerTimestamp,
+ TabletServerResource resource) {
this.rack = rack;
this.endpoints = endpoints;
this.registerTimestamp = registerTimestamp;
+ this.resource = resource;
}
public List<Endpoint> getEndpoints() {
@@ -53,6 +64,10 @@ public class TabletServerRegistration {
return rack;
}
+ public TabletServerResource getResource() {
+ return resource;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -64,12 +79,13 @@ public class TabletServerRegistration {
TabletServerRegistration that = (TabletServerRegistration) o;
return registerTimestamp == that.registerTimestamp
&& Objects.equals(endpoints, that.endpoints)
- && Objects.equals(rack, that.rack);
+ && Objects.equals(rack, that.rack)
+ && Objects.equals(resource, that.resource);
}
@Override
public int hashCode() {
- return Objects.hash(endpoints, registerTimestamp, rack);
+ return Objects.hash(endpoints, registerTimestamp, rack, resource);
}
@Override
@@ -81,6 +97,8 @@ public class TabletServerRegistration {
+ registerTimestamp
+ ", rack='"
+ rack
+ + "', resource="
+ + resource
+ '}';
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerde.java
index 976b2c431..389d84d10 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerde.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerde.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server.zk.data;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.server.metadata.TabletServerResource;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.fluss.utils.json.JsonDeserializer;
@@ -40,13 +41,15 @@ public class TabletServerRegistrationJsonSerde
new TabletServerRegistrationJsonSerde();
private static final String VERSION_KEY = "version";
- private static final int VERSION = 3;
+ private static final int VERSION = 4;
@Deprecated private static final String HOST = "host";
@Deprecated private static final String PORT = "port";
private static final String REGISTER_TIMESTAMP = "register_timestamp";
private static final String LISTENERS = "listeners";
private static final String RACK = "rack";
+ private static final String CPU_CORES = "cpu_cores";
+ private static final String MEMORY_BYTES = "memory_bytes";
@Override
public void serialize(
@@ -61,6 +64,13 @@ public class TabletServerRegistrationJsonSerde
if (tabletServerRegistration.getRack() != null) {
generator.writeStringField(RACK, tabletServerRegistration.getRack());
}
+ TabletServerResource resource = tabletServerRegistration.getResource();
+ if (resource.getCpuCores() != null) {
+ generator.writeNumberField(CPU_CORES, resource.getCpuCores());
+ }
+ if (resource.getMemoryBytes() != null) {
+ generator.writeNumberField(MEMORY_BYTES, resource.getMemoryBytes());
+ }
generator.writeEndObject();
}
@@ -69,6 +79,7 @@ public class TabletServerRegistrationJsonSerde
int version = node.get(VERSION_KEY).asInt();
List<Endpoint> endpoints;
String rack = null;
+ TabletServerResource resource = TabletServerResource.unknown();
if (version == 1) {
String host = node.get(HOST).asText();
int port = node.get(PORT).asInt();
@@ -80,9 +91,14 @@ public class TabletServerRegistrationJsonSerde
if (node.has(RACK)) {
rack = node.get(RACK).asText();
}
+ if (version >= VERSION) {
+ Double cpuCores = node.has(CPU_CORES) ? node.get(CPU_CORES).asDouble() : null;
+ Long memoryBytes = node.has(MEMORY_BYTES) ? node.get(MEMORY_BYTES).asLong() : null;
+ resource = new TabletServerResource(cpuCores, memoryBytes);
+ }
}
long registerTimestamp = node.get(REGISTER_TIMESTAMP).asLong();
- return new TabletServerRegistration(rack, endpoints, registerTimestamp);
+ return new TabletServerRegistration(rack, endpoints, registerTimestamp, resource);
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
index f8fadffb7..c7802413c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
@@ -19,6 +19,7 @@ package org.apache.fluss.server;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
import org.apache.fluss.config.cluster.AlterConfig;
import org.apache.fluss.config.cluster.AlterConfigOpType;
import org.apache.fluss.config.cluster.ConfigEntry;
@@ -62,6 +63,7 @@ import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link DynamicConfigManager}. */
@@ -567,6 +569,62 @@ public class DynamicConfigChangeTest {
assertThat(reconfiguredInterval.get()).isEqualTo(Duration.ofMinutes(5));
}
+ @Test
+ void testDynamicKvLeaderReplicaMemoryReservedChange() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, MemorySize.parse("8mb"));
+
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
+
+ AtomicReference<MemorySize> reconfiguredMemoryReserved = new AtomicReference<>();
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+ reconfiguredMemoryReserved.set(
+ newConfig.get(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ assertThatCode(
+ () ->
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions
+ .KV_LEADER_REPLICA_MEMORY_RESERVED
+ .key(),
+ "16mb",
+ AlterConfigOpType.SET))))
+ .doesNotThrowAnyException();
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED.key()))
+ .isEqualTo("16mb");
+ assertThat(reconfiguredMemoryReserved.get()).isEqualTo(MemorySize.parse("16mb"));
+
+ assertThatCode(
+ () ->
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions
+ .KV_LEADER_REPLICA_MEMORY_RESERVED
+ .key(),
+ "0b",
+ AlterConfigOpType.SET))))
+ .doesNotThrowAnyException();
+
+ zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED.key()))
+ .isEqualTo("0b");
+ assertThat(reconfiguredMemoryReserved.get()).isEqualTo(MemorySize.ZERO);
+ }
+
@Test
void testDynamicReconfigurationOfRemoteDataDirs() throws Exception {
Configuration configuration = new Configuration();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
index 9b4c8a20f..243b46ea0 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/AutoPartitionManagerTest.java
@@ -17,15 +17,21 @@
package org.apache.fluss.server.coordinator;
+import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.cluster.TabletServerInfo;
import org.apache.fluss.config.AutoPartitionTimeUnit;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metadata.TabletServerResource;
import org.apache.fluss.server.testutils.TestingServerMetadataCache;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -322,6 +328,7 @@ class AutoPartitionManagerTest {
new LakeCatalogDynamicLoader(new Configuration(), null, true)),
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -501,6 +508,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -580,6 +588,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -629,6 +638,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -696,6 +706,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
config,
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -721,6 +732,89 @@ class AutoPartitionManagerTest {
assertThat(partitionsNum).isEqualTo(5);
}
+ @Test
+ void testAutoCreatePartitionChecksCapacityWithoutReservation() throws Exception {
+ ZonedDateTime startTime =
+ LocalDateTime.parse("2025-04-26T00:00:00").atZone(ZoneId.systemDefault());
+ ManualClock clock = new ManualClock(startTime.toInstant().toEpochMilli());
+ ManuallyTriggeredScheduledExecutorService periodicExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ ReplicaCapacityController capacityController = capacityControllerWithCapacity(16);
+ AutoPartitionManager autoPartitionManager =
+ new AutoPartitionManager(
+ new TestingServerMetadataCache(3),
+ metadataManager,
+ remoteDirDynamicLoader,
+ new Configuration(),
+ capacityController,
+ clock,
+ periodicExecutor);
+ autoPartitionManager.start();
+
+ TableInfo table = createPartitionedTable(-1, 2, AutoPartitionTimeUnit.HOUR);
+ TablePath tablePath = table.getTablePath();
+ autoPartitionManager.addAutoPartitionTable(table, true);
+ periodicExecutor.triggerNonPeriodicScheduledTask();
+
+ Map<String, PartitionRegistration> partitions =
+ zookeeperClient.getPartitionRegistrations(tablePath);
+ // This unit test has no coordinator event thread to publish the created partition buckets
+ // back as observed state, so both best-effort checks pass without reserving capacity.
+ assertThat(partitions.keySet()).containsExactlyInAnyOrder("2025042600", "2025042601");
+ assertThat(capacityController.getKvLeaderReplicaCount()).isZero();
+ }
+
+ @Test
+ void testAutoDropPartitionDoesNotMutateObservedKvLeaderReplicaCount() throws Exception {
+ ZonedDateTime startTime =
+ LocalDateTime.parse("2025-04-26T00:00:00").atZone(ZoneId.systemDefault());
+ ManualClock clock = new ManualClock(startTime.toInstant().toEpochMilli());
+ ManuallyTriggeredScheduledExecutorService periodicExecutor =
+ new ManuallyTriggeredScheduledExecutorService();
+ ReplicaCapacityController capacityController = capacityControllerWithCapacity(64);
+ AutoPartitionManager autoPartitionManager =
+ new AutoPartitionManager(
+ new TestingServerMetadataCache(3),
+ metadataManager,
+ remoteDirDynamicLoader,
+ new Configuration(),
+ capacityController,
+ clock,
+ periodicExecutor);
+ autoPartitionManager.start();
+
+ TableInfo table = createPartitionedTable(1, 0, AutoPartitionTimeUnit.HOUR);
+ TablePath tablePath = table.getTablePath();
+ autoPartitionManager.addAutoPartitionTable(table, false);
+ PartitionAssignment partitionAssignment = partitionAssignment(table);
+ metadataManager.createPartition(
+ tablePath,
+ table.getTableId(),
+ remoteDataDir,
+ partitionAssignment,
+ fromPartitionName(table.getPartitionKeys(), "2025042600"),
+ false);
+ metadataManager.createPartition(
+ tablePath,
+ table.getTableId(),
+ remoteDataDir,
+ partitionAssignment,
+ fromPartitionName(table.getPartitionKeys(), "2025042601"),
+ false);
+ autoPartitionManager.addPartition(table.getTableId(), "2025042600");
+ autoPartitionManager.addPartition(table.getTableId(), "2025042601");
+ capacityController.updateObservedKvLeaderReplicaCount((long) table.getNumBuckets() * 2);
+
+ clock.advanceTime(Duration.ofHours(2));
+ periodicExecutor.triggerPeriodicScheduledTasks();
+
+ Map<String, PartitionRegistration> partitions =
+ zookeeperClient.getPartitionRegistrations(tablePath);
+ assertThat(partitions.keySet()).containsExactly("2025042601");
+ assertThat(capacityController.getKvLeaderReplicaCount())
+ .isEqualTo((long) table.getNumBuckets() * 2);
+ }
+
@Test
void testUpdateAutoPartitionNumRetention() throws Exception {
// Start at a well-known time
@@ -737,6 +831,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -784,6 +879,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -844,6 +940,7 @@ class AutoPartitionManagerTest {
metadataManager,
remoteDirDynamicLoader,
new Configuration(),
+ disabledCapacityController(),
clock,
periodicExecutor);
autoPartitionManager.start();
@@ -1225,4 +1322,42 @@ class AutoPartitionManagerTest {
original.getCreatedTime(),
System.currentTimeMillis());
}
+
+ private static ReplicaCapacityController capacityControllerWithCapacity(long capacity) {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ Collections.singleton(
+ new ServerInfo(
+ 0,
+ null,
+ Endpoint.fromListenersString("INTERNAL://localhost:10000"),
+ ServerType.TABLET_SERVER,
+ new TabletServerResource(null, capacity))),
+ Collections.emptyMap());
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, new MemorySize(1));
+ return new ReplicaCapacityController(conf, metadataCache);
+ }
+
+ private static ReplicaCapacityController disabledCapacityController() {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, MemorySize.ZERO);
+ return new ReplicaCapacityController(conf, new CoordinatorMetadataCache());
+ }
+
+ private static PartitionAssignment partitionAssignment(TableInfo table) {
+ int replicaFactor = table.getTableConfig().getReplicationFactor();
+ Map<Integer, BucketAssignment> bucketAssignments =
+ generateAssignment(
+ table.getNumBuckets(),
+ replicaFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ })
+ .getBucketAssignments();
+ return new PartitionAssignment(table.getTableId(), bucketAssignments);
+ }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
index 557d75a6a..6933ccb8c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
@@ -46,6 +46,26 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link CoordinatorContext}. */
class CoordinatorContextTest {
+ @Test
+ void testKvBucketsAreTrackedIdempotently() {
+ CoordinatorContext context = new CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket first = new TableBucket(1L, 0);
+ TableBucket second = new TableBucket(1L, 1);
+
+ context.addKvBuckets(Arrays.asList(first, second));
+ context.addKvBuckets(Arrays.asList(first, second));
+
+ assertThat(context.getKvBucketCount()).isEqualTo(2);
+
+ context.removeKvBuckets(Collections.singleton(first));
+ context.removeKvBuckets(Collections.singleton(first));
+
+ assertThat(context.getKvBucketCount()).isEqualTo(1);
+
+ context.resetContext();
+ assertThat(context.getKvBucketCount()).isZero();
+ }
+
@Test
void testGetLakeTableCount() {
CoordinatorContext context = new CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index d3558449f..5a5a68b3c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -88,6 +88,7 @@ import org.apache.fluss.server.zk.data.ZkData;
import org.apache.fluss.server.zk.data.ZkData.PartitionIdsZNode;
import org.apache.fluss.server.zk.data.ZkData.TableIdsZNode;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
+import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService;
import org.apache.fluss.types.DataTypes;
import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.clock.SystemClock;
@@ -172,6 +173,7 @@ class CoordinatorEventProcessorTest {
private LakeTableTieringManager lakeTableTieringManager;
private CompletedSnapshotStoreManager completedSnapshotStoreManager;
private CoordinatorMetadataCache serverMetadataCache;
+ private ReplicaCapacityController replicaCapacityController;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
private Scheduler scheduler;
private String remoteDataDir;
@@ -216,12 +218,14 @@ class CoordinatorEventProcessorTest {
remoteDataDir = zookeeperClient.getDefaultRemoteDataDir();
Configuration conf = new Configuration();
conf.setString(ConfigOptions.REMOTE_DATA_DIR, remoteDataDir);
+ replicaCapacityController = new ReplicaCapacityController(conf, serverMetadataCache);
autoPartitionManager =
new AutoPartitionManager(
serverMetadataCache,
metadataManager,
new RemoteDirDynamicLoader(conf),
- new Configuration());
+ conf,
+ replicaCapacityController);
kvSnapshotLeaseManager =
new KvSnapshotLeaseManager(
Duration.ofMinutes(10).toMillis(),
@@ -255,6 +259,140 @@ class CoordinatorEventProcessorTest {
ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupPath(PartitionIdsZNode.path());
}
+ @Test
+ void testLoadedAssignmentsTrackKnownKvAndUnknownTablesConservatively() throws Exception {
+ long kvTableId = 10001L;
+ long logTableId = 10002L;
+ long orphanTableId = 10003L;
+ TableDescriptor logTable =
+ TableDescriptor.builder()
+ .schema(Schema.newBuilder().column("a", DataTypes.INT()).build())
+ .distributedBy(1)
+ .build();
+ TableBucket kvBucket = new TableBucket(kvTableId, 0);
+ TableBucket logBucket = new TableBucket(logTableId, 0);
+ TableBucket orphanBucket = new TableBucket(orphanTableId, 0);
+
+ fromCtx(
+ ctx -> {
+ long now = System.currentTimeMillis();
+ ctx.putTableInfo(
+ TableInfo.of(
+ TablePath.of(defaultDatabase, "known_kv"),
+ kvTableId,
+ 0,
+ TEST_TABLE,
+ remoteDataDir,
+ now,
+ now));
+ ctx.putTableInfo(
+ TableInfo.of(
+ TablePath.of(defaultDatabase, "known_log"),
+ logTableId,
+ 0,
+ logTable,
+ remoteDataDir,
+ now,
+ now));
+
+ eventProcessor.trackKvBucketsForLoadedAssignment(
+ kvTableId, Collections.singleton(kvBucket));
+ eventProcessor.trackKvBucketsForLoadedAssignment(
+ logTableId, Collections.singleton(logBucket));
+ eventProcessor.trackKvBucketsForLoadedAssignment(
+ orphanTableId, Collections.singleton(orphanBucket));
+ return null;
+ });
+
+ assertThat(fromCtx(CoordinatorContext::getKvBucketCount)).isEqualTo(2);
+ fromCtx(
+ ctx -> {
+ ctx.removeKvBuckets(Arrays.asList(kvBucket, logBucket, orphanBucket));
+ return null;
+ });
+ assertThat(fromCtx(CoordinatorContext::getKvBucketCount)).isZero();
+ }
+
+ @Test
+ void testStartupAndShutdownPublishObservedKvLeaderReplicaCount() throws Exception {
+ initCoordinatorChannel();
+ TablePath tablePath = TablePath.of(defaultDatabase, "startup_observed_count");
+ TableAssignment tableAssignment =
+ generateAssignment(
+ N_BUCKETS,
+ REPLICATION_FACTOR,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ long tableId =
+ metadataManager.createTable(
+ tablePath, remoteDataDir, TEST_TABLE, tableAssignment, false);
+ verifyTableCreated(tableId, tableAssignment, N_BUCKETS, REPLICATION_FACTOR);
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isEqualTo(N_BUCKETS);
+
+ eventProcessor.shutdown();
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isZero();
+
+ // Simulate a stale value left by a previous CoordinatorContext. Startup must replace it
+ // with the count reconstructed from the loaded assignments.
+ replicaCapacityController.updateObservedKvLeaderReplicaCount(100);
+ eventProcessor = buildCoordinatorEventProcessor();
+ initCoordinatorChannel();
+ eventProcessor.startup();
+
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isEqualTo(N_BUCKETS);
+ }
+
+ @Test
+ void testAutoPartitionInitializationAfterCapacityInputsRestored() throws Exception {
+ initCoordinatorChannel();
+ TablePath kvTablePath = TablePath.of(defaultDatabase, "startup_capacity_kv");
+ TableAssignment tableAssignment =
+ generateAssignment(
+ N_BUCKETS,
+ REPLICATION_FACTOR,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ long kvTableId =
+ metadataManager.createTable(
+ kvTablePath, remoteDataDir, TEST_TABLE, tableAssignment, false);
+ verifyTableCreated(kvTableId, tableAssignment, N_BUCKETS, REPLICATION_FACTOR);
+
+ TablePath autoPartitionTablePath =
+ TablePath.of(defaultDatabase, "startup_capacity_auto_partition");
+ long autoPartitionTableId =
+ metadataManager.createTable(
+ autoPartitionTablePath, remoteDataDir, getPartitionedTable(), null, false);
+
+ eventProcessor.shutdown();
+ autoPartitionManager.close();
+
+ serverMetadataCache = new CoordinatorMetadataCache();
+ replicaCapacityController =
+ new ReplicaCapacityController(new Configuration(), serverMetadataCache);
+ RecordingAutoPartitionManager recordingAutoPartitionManager =
+ new RecordingAutoPartitionManager(
+ serverMetadataCache,
+ metadataManager,
+ remoteDataDir,
+ replicaCapacityController);
+ autoPartitionManager = recordingAutoPartitionManager;
+ eventProcessor = buildCoordinatorEventProcessor();
+ initCoordinatorChannel();
+ eventProcessor.startup();
+
+ assertThat(recordingAutoPartitionManager.getInitializedTableIds())
+ .contains(autoPartitionTableId);
+ assertThat(recordingAutoPartitionManager.getObservedKvLeaderReplicaCountAtInit())
+ .isEqualTo(N_BUCKETS);
+ assertThat(recordingAutoPartitionManager.getLiveTabletServerCountAtInit()).isEqualTo(3);
+ }
+
@Test
void testCreateAndDropTable() throws Exception {
// make sure all request to gateway should be successful
@@ -283,6 +421,8 @@ class CoordinatorEventProcessorTest {
t2, remoteDataDir, tableDescriptor, tableAssignment, false);
verifyTableCreated(t2Id, tableAssignment, nBuckets, replicationFactor);
+ retryVerifyContext(ctx -> assertThat(ctx.getKvBucketCount()).isEqualTo(6));
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isEqualTo(6);
// mock CompletedSnapshotStore
for (TableBucket tableBucket : allTableBuckets(t1Id, nBuckets)) {
@@ -295,6 +435,8 @@ class CoordinatorEventProcessorTest {
metadataManager.dropTable(t1, false);
verifyTableDropped(t1Id);
+ retryVerifyContext(ctx -> assertThat(ctx.getKvBucketCount()).isEqualTo(3));
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isEqualTo(3);
// verify CompleteSnapshotStore has been removed when the table is dropped
assertThat(completedSnapshotStoreManager.getBucketCompletedSnapshotStores()).isEmpty();
@@ -326,6 +468,8 @@ class CoordinatorEventProcessorTest {
Set<TableBucketReplica> tableBucketReplicas =
fromCtx(ctx -> ctx.getAllReplicasForTable(t2Id));
assertThat(tableBucketReplicas).isEmpty();
+ assertThat(fromCtx(CoordinatorContext::getKvBucketCount)).isZero();
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isZero();
}
@Test
@@ -1950,6 +2094,7 @@ class CoordinatorEventProcessorTest {
serverMetadataCache,
testCoordinatorChannelManager,
new CoordinatorContext(zkEpoch),
+ replicaCapacityController,
autoPartitionManager,
lakeTableTieringManager,
TestingMetricGroups.COORDINATOR_METRICS,
@@ -1961,6 +2106,57 @@ class CoordinatorEventProcessorTest {
SystemClock.getInstance());
}
+ private static class RecordingAutoPartitionManager extends AutoPartitionManager {
+
+ private final CoordinatorMetadataCache metadataCache;
+ private final ReplicaCapacityController replicaCapacityController;
+
+ private Set<Long> initializedTableIds = Collections.emptySet();
+ private long observedKvLeaderReplicaCountAtInit = -1;
+ private int liveTabletServerCountAtInit = -1;
+
+ private RecordingAutoPartitionManager(
+ CoordinatorMetadataCache metadataCache,
+ MetadataManager metadataManager,
+ String remoteDataDir,
+ ReplicaCapacityController replicaCapacityController) {
+ super(
+ metadataCache,
+ metadataManager,
+ new RemoteDirDynamicLoader(
+ Configuration.fromMap(
+ Collections.singletonMap(
+ ConfigOptions.REMOTE_DATA_DIR.key(), remoteDataDir))),
+ new Configuration(),
+ replicaCapacityController,
+ SystemClock.getInstance(),
+ new ManuallyTriggeredScheduledExecutorService());
+ this.metadataCache = metadataCache;
+ this.replicaCapacityController = replicaCapacityController;
+ }
+
+ @Override
+ public void initAutoPartitionTables(List<TableInfo> tableInfos) {
+ initializedTableIds =
+ tableInfos.stream().map(TableInfo::getTableId).collect(Collectors.toSet());
+ observedKvLeaderReplicaCountAtInit =
+ replicaCapacityController.getKvLeaderReplicaCount();
+ liveTabletServerCountAtInit = metadataCache.getLiveTabletServerInfos().size();
+ }
+
+ private Set<Long> getInitializedTableIds() {
+ return initializedTableIds;
+ }
+
+ private long getObservedKvLeaderReplicaCountAtInit() {
+ return observedKvLeaderReplicaCountAtInit;
+ }
+
+ private int getLiveTabletServerCountAtInit() {
+ return liveTabletServerCountAtInit;
+ }
+ }
+
private void initCoordinatorChannel() throws Exception {
makeSendLeaderAndStopRequestAlwaysSuccess(
testCoordinatorChannelManager,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ReplicaCapacityControllerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ReplicaCapacityControllerTest.java
new file mode 100644
index 000000000..b93808777
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ReplicaCapacityControllerTest.java
@@ -0,0 +1,224 @@
+/*
+ * 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.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.cluster.rebalance.ServerTag;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.registry.NOPMetricRegistry;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metadata.TabletServerResource;
+import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link ReplicaCapacityController}. */
+class ReplicaCapacityControllerTest {
+
+ @Test
+ void testCapacityUsesAverageMemoryForUnknownTabletServers() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(
+ Arrays.asList(
+ tabletServer(0, 100L),
+ tabletServer(1, 100L),
+ unknownMemoryTabletServer(2),
+ tabletServer(3, 10000L))),
+ Collections.singletonMap(3, ServerTag.PERMANENT_OFFLINE));
+
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(configWithMemoryReserved(10), metadataCache);
+
+ assertThat(controller.getKvLeaderReplicaCapacity()).isEqualTo(30);
+ }
+
+ @Test
+ void testRegistersKvLeaderReplicaMetrics() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(Arrays.asList(tabletServer(0, 100L), tabletServer(1, 100L))),
+ Collections.emptyMap());
+ CoordinatorMetricGroup metricGroup =
+ new CoordinatorMetricGroup(NOPMetricRegistry.INSTANCE, "cluster", "localhost", "0");
+
+ try {
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(
+ configWithMemoryReserved(10), metadataCache, metricGroup);
+ controller.updateObservedKvLeaderReplicaCount(3);
+
+ Gauge<?> countGauge =
+ (Gauge<?>) metricGroup.getMetrics().get(MetricNames.KV_LEADER_REPLICA_COUNT);
+ Gauge<?> capacityGauge =
+ (Gauge<?>) metricGroup.getMetrics().get(MetricNames.KV_LEADER_REPLICA_CAPACITY);
+ assertThat(countGauge.getValue()).isEqualTo(3L);
+ assertThat(capacityGauge.getValue()).isEqualTo(20L);
+ } finally {
+ metricGroup.close();
+ }
+ }
+
+ @Test
+ void testCapacityDisabledIfKnownMemoryTabletServersAreNotMajority() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(
+ Arrays.asList(
+ tabletServer(0, 100L),
+ unknownMemoryTabletServer(1),
+ unknownMemoryTabletServer(2))),
+ Collections.emptyMap());
+
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(configWithMemoryReserved(10), metadataCache);
+
+ assertThat(controller.getKvLeaderReplicaCapacity())
+ .isEqualTo(ReplicaCapacityController.CAPACITY_LIMIT_DISABLED);
+ }
+
+ @Test
+ void testCapacityControlDisabledByDefault() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(Arrays.asList(tabletServer(0, 100L), tabletServer(1, 100L))),
+ Collections.emptyMap());
+
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(new Configuration(), metadataCache);
+
+ assertThat(controller.getKvLeaderReplicaMemoryReservedBytes()).isZero();
+ assertThat(controller.getKvLeaderReplicaCapacity())
+ .isEqualTo(ReplicaCapacityController.CAPACITY_LIMIT_DISABLED);
+ controller.checkCanCreateKvLeaderReplicas(201);
+ assertThat(controller.getKvLeaderReplicaCount()).isZero();
+ }
+
+ @Test
+ void testAdmissionCheckUsesObservedCountWithoutReservation() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(
+ Arrays.asList(
+ tabletServer(0, 100L),
+ tabletServer(1, 100L),
+ unknownMemoryTabletServer(2))),
+ Collections.emptyMap());
+
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(configWithMemoryReserved(10), metadataCache);
+ controller.updateObservedKvLeaderReplicaCount(20);
+ controller.checkCanCreateKvLeaderReplicas(10);
+
+ assertThat(controller.getKvLeaderReplicaCount()).isEqualTo(20);
+ assertThatThrownBy(() -> controller.checkCanCreateKvLeaderReplicas(11))
+ .isInstanceOf(InsufficientKvLeaderReplicaCapacityException.class)
+ .hasMessageContaining("observedKvLeaderReplicaCount=20")
+ .hasMessageContaining("requestedKvLeaderReplicaCount=11")
+ .hasMessageContaining("kvLeaderReplicaCapacity=30");
+ assertThat(controller.getKvLeaderReplicaCount()).isEqualTo(20);
+ }
+
+ @Test
+ void testAdmissionUsesLatestMemoryReservedConfig() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(Arrays.asList(tabletServer(0, 100L), tabletServer(1, 100L))),
+ Collections.emptyMap());
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(configWithMemoryReserved(10), metadataCache);
+ controller.updateObservedKvLeaderReplicaCount(15);
+
+ controller.checkCanCreateKvLeaderReplicas(5);
+ controller.reconfigure(configWithMemoryReserved(20));
+
+ assertThatThrownBy(() -> controller.checkCanCreateKvLeaderReplicas(1))
+ .isInstanceOf(InsufficientKvLeaderReplicaCapacityException.class)
+ .hasMessageContaining("kvLeaderReplicaCapacity=10");
+ }
+
+ @Test
+ void testReconfigureMemoryReserved() {
+ CoordinatorMetadataCache metadataCache = new CoordinatorMetadataCache();
+ metadataCache.updateMetadata(
+ null,
+ new HashSet<>(Arrays.asList(tabletServer(0, 100L), tabletServer(1, 100L))),
+ Collections.emptyMap());
+ ReplicaCapacityController controller =
+ new ReplicaCapacityController(configWithMemoryReserved(10), metadataCache);
+
+ controller.reconfigure(configWithMemoryReserved(20));
+
+ assertThat(controller.getKvLeaderReplicaMemoryReservedBytes()).isEqualTo(20);
+ assertThat(controller.getKvLeaderReplicaCapacity()).isEqualTo(10);
+
+ Configuration disabledConfig = configWithMemoryReserved(0);
+ controller.validate(disabledConfig);
+ controller.reconfigure(disabledConfig);
+ assertThat(controller.getKvLeaderReplicaMemoryReservedBytes()).isZero();
+ assertThat(controller.getKvLeaderReplicaCapacity())
+ .isEqualTo(ReplicaCapacityController.CAPACITY_LIMIT_DISABLED);
+
+ controller.reconfigure(configWithMemoryReserved(10));
+ assertThat(controller.getKvLeaderReplicaCapacity()).isEqualTo(20);
+ }
+
+ private static Configuration configWithMemoryReserved(long bytes) {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, new MemorySize(bytes));
+ return conf;
+ }
+
+ private static ServerInfo tabletServer(int serverId, long memoryBytes) {
+ return new ServerInfo(
+ serverId,
+ null,
+ Endpoint.fromListenersString("INTERNAL://localhost:" + (10000 + serverId)),
+ ServerType.TABLET_SERVER,
+ new TabletServerResource(null, memoryBytes));
+ }
+
+ private static ServerInfo unknownMemoryTabletServer(int serverId) {
+ return new ServerInfo(
+ serverId,
+ null,
+ Endpoint.fromListenersString("INTERNAL://localhost:" + (10000 + serverId)),
+ ServerType.TABLET_SERVER,
+ TabletServerResource.unknown());
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
index 0428d4e10..78684f1c8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
@@ -23,15 +23,18 @@ import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.AutoPartitionTimeUnit;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
import org.apache.fluss.exception.DatabaseAlreadyExistException;
import org.apache.fluss.exception.DatabaseNotEmptyException;
import org.apache.fluss.exception.DatabaseNotExistException;
+import org.apache.fluss.exception.InsufficientKvLeaderReplicaCapacityException;
import org.apache.fluss.exception.InvalidDatabaseException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.PartitionNotExistException;
import org.apache.fluss.exception.SchemaNotExistException;
import org.apache.fluss.exception.TableAlreadyExistException;
import org.apache.fluss.exception.TableNotExistException;
+import org.apache.fluss.metadata.PartitionSpec;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
@@ -95,11 +98,13 @@ import java.util.stream.Collectors;
import static org.apache.fluss.config.ConfigOptions.CURRENT_KV_FORMAT_VERSION;
import static org.apache.fluss.config.ConfigOptions.DEFAULT_LISTENER_NAME;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createPartition;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newAlterTableRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newCreateDatabaseRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newCreateTableRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDatabaseExistsRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDropDatabaseRequest;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDropPartitionRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDropTableRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newGetTableInfoRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newListTablesRequest;
@@ -842,6 +847,10 @@ class TableManagerITCase {
.build();
}
+ private static PartitionSpec partitionSpec(String dt) {
+ return new PartitionSpec(Collections.singletonMap("dt", dt));
+ }
+
private static List<PbAddColumn> alterTableAddColumns() {
List<PbAddColumn> addColumns = new ArrayList<>();
PbAddColumn newNestedColumn = new PbAddColumn();
@@ -915,6 +924,166 @@ class TableManagerITCase {
// Test methods for table creation restrictions
+ @Test
+ void testKvLeaderReplicaCapacityRestriction() throws Exception {
+ Configuration conf = initConf();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, new MemorySize(1));
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE, new MemorySize(3));
+ FlussClusterExtension limitedCluster =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(1)
+ .setClusterConf(conf)
+ .build();
+
+ try {
+ limitedCluster.start();
+
+ AdminGateway gateway = limitedCluster.newCoordinatorClient();
+ TablePath firstTablePath = TablePath.of("fluss", "kv_capacity_first");
+ TablePath secondTablePath = TablePath.of("fluss", "kv_capacity_second");
+ gateway.createTable(newCreateTableRequest(firstTablePath, newPkTable(), false)).get();
+ waitForKvLeaderReplicaCount(limitedCluster, 3);
+
+ assertThatThrownBy(
+ () ->
+ gateway.createTable(
+ newCreateTableRequest(
+ secondTablePath,
+ newPkTable().withBucketCount(1),
+ false))
+ .get())
+ .cause()
+ .isInstanceOf(InsufficientKvLeaderReplicaCapacityException.class)
+ .hasMessageContaining("observedKvLeaderReplicaCount=3")
+ .hasMessageContaining("requestedKvLeaderReplicaCount=1")
+ .hasMessageContaining("kvLeaderReplicaCapacity=3");
+
+ gateway.dropTable(newDropTableRequest("fluss", "kv_capacity_first", false)).get();
+ waitForKvLeaderReplicaCount(limitedCluster, 0);
+ gateway.createTable(
+ newCreateTableRequest(
+ secondTablePath, newPkTable().withBucketCount(1), false))
+ .get();
+
+ assertThat(gateway.tableExists(newTableExistsRequest(secondTablePath)).get().isExists())
+ .isTrue();
+ } finally {
+ limitedCluster.close();
+ }
+ }
+
+ @Test
+ void testKvLeaderReplicaCapacityRestrictionForPartitionedTable() throws Exception {
+ Configuration conf = initConf();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, new MemorySize(1));
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE, new MemorySize(3));
+ FlussClusterExtension limitedCluster =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(1)
+ .setClusterConf(conf)
+ .build();
+
+ try {
+ limitedCluster.start();
+
+ AdminGateway gateway = limitedCluster.newCoordinatorClient();
+ TablePath tablePath = TablePath.of("fluss", "partitioned_kv_capacity");
+ TableDescriptor tableDescriptor =
+ newPartitionedTableBuilder(null)
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED.key(), "false")
+ .build();
+ gateway.createTable(newCreateTableRequest(tablePath, tableDescriptor, false)).get();
+
+ PartitionSpec firstPartition = partitionSpec("20260707");
+ PartitionSpec secondPartition = partitionSpec("20260708");
+ createPartition(limitedCluster, tablePath, firstPartition, false);
+ waitForKvLeaderReplicaCount(limitedCluster, 3);
+
+ assertThatThrownBy(
+ () ->
+ createPartition(
+ limitedCluster, tablePath, secondPartition, false))
+ .cause()
+ .isInstanceOf(InsufficientKvLeaderReplicaCapacityException.class)
+ .hasMessageContaining("observedKvLeaderReplicaCount=3")
+ .hasMessageContaining("requestedKvLeaderReplicaCount=3")
+ .hasMessageContaining("kvLeaderReplicaCapacity=3");
+
+ gateway.dropPartition(newDropPartitionRequest(tablePath, firstPartition, false)).get();
+ waitForKvLeaderReplicaCount(limitedCluster, 0);
+ createPartition(limitedCluster, tablePath, secondPartition, false);
+
+ assertThat(limitedCluster.getZooKeeperClient().getPartition(tablePath, "20260708"))
+ .isPresent();
+ } finally {
+ limitedCluster.close();
+ }
+ }
+
+ @Test
+ void testKvLeaderReplicaCapacityRecomputedOnTabletServerReRegistration() throws Exception {
+ Configuration conf = initConf();
+ conf.set(ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED, new MemorySize(1));
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE, new MemorySize(3));
+ FlussClusterExtension cluster =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(3)
+ .setClusterConf(conf)
+ .build();
+
+ try {
+ cluster.start();
+
+ AdminGateway gateway = cluster.newCoordinatorClient();
+ // 3 live tablet servers, each advertising 3 bytes with 1 byte reserved -> capacity 9.
+ waitForKvLeaderReplicaCapacity(cluster, 9);
+
+ TablePath tablePath = TablePath.of("fluss", "kv_capacity_reregister");
+ gateway.createTable(newCreateTableRequest(tablePath, newPkTable(), false)).get();
+ // newPkTable() has 3 buckets -> 3 observed KV leader replicas.
+ waitForKvLeaderReplicaCount(cluster, 3);
+
+ // Bring one tablet server down: capacity shrinks to reflect the live servers, while the
+ // observed KV leader count is unaffected by tablet server churn (leaders may migrate,
+ // but the number of KV buckets tracked by the coordinator stays the same).
+ cluster.stopTabletServer(1);
+ waitForKvLeaderReplicaCapacity(cluster, 6);
+ waitForKvLeaderReplicaCount(cluster, 3);
+
+ // Re-register the tablet server: capacity is recomputed from the live servers again and
+ // the observed KV leader count remains stable.
+ cluster.startTabletServer(1);
+ waitForKvLeaderReplicaCapacity(cluster, 9);
+ waitForKvLeaderReplicaCount(cluster, 3);
+ } finally {
+ cluster.close();
+ }
+ }
+
+ private static void waitForKvLeaderReplicaCount(
+ FlussClusterExtension cluster, long expectedCount) {
+ retry(
+ Duration.ofSeconds(30),
+ () ->
+ assertThat(
+ cluster.getCoordinatorServer()
+ .getReplicaCapacityController()
+ .getKvLeaderReplicaCount())
+ .isEqualTo(expectedCount));
+ }
+
+ private static void waitForKvLeaderReplicaCapacity(
+ FlussClusterExtension cluster, long expectedCapacity) {
+ retry(
+ Duration.ofSeconds(30),
+ () ->
+ assertThat(
+ cluster.getCoordinatorServer()
+ .getReplicaCapacityController()
+ .getKvLeaderReplicaCapacity())
+ .isEqualTo(expectedCapacity));
+ }
+
@Test
void testLogTableCreationRestriction() throws Exception {
// Test with cluster that disallows log table creation
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
index 68b65d80d..8116cfcb3 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
@@ -30,6 +30,7 @@ import org.apache.fluss.server.coordinator.event.TestingEventManager;
import org.apache.fluss.server.coordinator.statemachine.ReplicaStateMachine;
import org.apache.fluss.server.coordinator.statemachine.TableBucketStateMachine;
import org.apache.fluss.server.entity.DeleteReplicaResultForBucket;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
import org.apache.fluss.server.metadata.ServerInfo;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZkEpoch;
@@ -85,6 +86,7 @@ class TableManagerTest {
private static ExecutorService ioExecutor;
private CoordinatorContext coordinatorContext;
+ private ReplicaCapacityController replicaCapacityController;
private TableManager tableManager;
private TableLifecycleThrottler lifecycleThrottler;
private TestingEventManager testingEventManager;
@@ -123,6 +125,8 @@ class TableManagerTest {
private void initTableManager() {
testingEventManager = new TestingEventManager();
coordinatorContext = new CoordinatorContext(zkEpoch);
+ replicaCapacityController =
+ new ReplicaCapacityController(new Configuration(), new CoordinatorMetadataCache());
testCoordinatorChannelManager = new TestCoordinatorChannelManager();
Configuration conf = new Configuration();
conf.setString(ConfigOptions.REMOTE_DATA_DIR, "/tmp/fluss/remote-data");
@@ -147,6 +151,7 @@ class TableManagerTest {
new TableManager(
metadataManager,
coordinatorContext,
+ replicaCapacityController,
replicaStateMachine,
tableBucketStateMachine,
new RemoteStorageCleaner(conf, ioExecutor),
@@ -181,6 +186,8 @@ class TableManagerTest {
System.currentTimeMillis()));
tableManager.onCreateNewTable(DATA1_TABLE_PATH, tableId, assignment);
+ assertThat(coordinatorContext.getKvBucketCount()).isZero();
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isZero();
// all replica should be online
checkReplicaOnline(tableId, null, assignment);
// clear the assignment for the table
@@ -204,10 +211,15 @@ class TableManagerTest {
System.currentTimeMillis(),
System.currentTimeMillis()));
tableManager.onCreateNewTable(DATA1_TABLE_PATH_PK, tableId, assignment);
+ assertThat(coordinatorContext.getKvBucketCount()).isEqualTo(assignment.getBuckets().size());
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount())
+ .isEqualTo(assignment.getBuckets().size());
// now, delete the created table
coordinatorContext.queueTableDeletion(Collections.singleton(tableId));
tableManager.onDeleteTable(tableId);
+ assertThat(coordinatorContext.getKvBucketCount()).isZero();
+ assertThat(replicaCapacityController.getKvLeaderReplicaCount()).isZero();
// make sure the delete replica success events in event manager is equal to the expected
checkReplicaDelete(tableId, null, assignment);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
index dfe335fbe..57731571a 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
@@ -29,6 +29,7 @@ import org.apache.fluss.server.coordinator.CoordinatorEventProcessor;
import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
import org.apache.fluss.server.coordinator.LakeTableTieringManager;
import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.coordinator.ReplicaCapacityController;
import org.apache.fluss.server.coordinator.TestCoordinatorChannelManager;
import org.apache.fluss.server.coordinator.event.CoordinatorEvent;
import org.apache.fluss.server.coordinator.event.EventManager;
@@ -83,6 +84,7 @@ public class RebalanceManagerTest {
private CoordinatorMetadataCache serverMetadataCache;
private TestCoordinatorChannelManager testCoordinatorChannelManager;
private AutoPartitionManager autoPartitionManager;
+ private ReplicaCapacityController replicaCapacityController;
private LakeTableTieringManager lakeTableTieringManager;
private RebalanceManager rebalanceManager;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
@@ -117,12 +119,16 @@ public class RebalanceManagerTest {
scheduler = new FlussScheduler(1);
scheduler.startup();
+ replicaCapacityController =
+ new ReplicaCapacityController(
+ conf, serverMetadataCache, TestingMetricGroups.COORDINATOR_METRICS);
autoPartitionManager =
new AutoPartitionManager(
serverMetadataCache,
metadataManager,
new RemoteDirDynamicLoader(conf),
- conf);
+ conf,
+ replicaCapacityController);
lakeTableTieringManager =
new LakeTableTieringManager(TestingMetricGroups.LAKE_TIERING_METRICS);
CoordinatorEventProcessor eventProcessor = buildCoordinatorEventProcessor(conf);
@@ -314,6 +320,7 @@ public class RebalanceManagerTest {
serverMetadataCache,
testCoordinatorChannelManager,
new CoordinatorContext(zkEpoch),
+ replicaCapacityController,
autoPartitionManager,
lakeTableTieringManager,
TestingMetricGroups.COORDINATOR_METRICS,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
index 630128546..b12e245fe 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
@@ -33,6 +33,7 @@ import org.apache.fluss.server.coordinator.CoordinatorTestUtils;
import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
import org.apache.fluss.server.coordinator.LakeTableTieringManager;
import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.coordinator.ReplicaCapacityController;
import org.apache.fluss.server.coordinator.TestCoordinatorChannelManager;
import org.apache.fluss.server.coordinator.event.CoordinatorEventManager;
import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
@@ -94,6 +95,7 @@ class TableBucketStateMachineTest {
private TestCoordinatorChannelManager testCoordinatorChannelManager;
private CoordinatorRequestBatch coordinatorRequestBatch;
private AutoPartitionManager autoPartitionManager;
+ private ReplicaCapacityController replicaCapacityController;
private LakeTableTieringManager lakeTableTieringManager;
private CoordinatorMetadataCache serverMetadataCache;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
@@ -124,6 +126,9 @@ class TableBucketStateMachineTest {
},
coordinatorContext);
serverMetadataCache = new CoordinatorMetadataCache();
+ replicaCapacityController =
+ new ReplicaCapacityController(
+ conf, serverMetadataCache, TestingMetricGroups.COORDINATOR_METRICS);
autoPartitionManager =
new AutoPartitionManager(
serverMetadataCache,
@@ -132,7 +137,8 @@ class TableBucketStateMachineTest {
new Configuration(),
new LakeCatalogDynamicLoader(new Configuration(), null, true)),
new RemoteDirDynamicLoader(conf),
- new Configuration());
+ conf,
+ replicaCapacityController);
lakeTableTieringManager =
new LakeTableTieringManager(TestingMetricGroups.LAKE_TIERING_METRICS);
@@ -316,6 +322,7 @@ class TableBucketStateMachineTest {
new Configuration(),
TestingClientMetricGroup.newInstance())),
coordinatorContext,
+ replicaCapacityController,
autoPartitionManager,
lakeTableTieringManager,
TestingMetricGroups.COORDINATOR_METRICS,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerResourceProbeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerResourceProbeTest.java
new file mode 100644
index 000000000..e12c36578
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerResourceProbeTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.tablet;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.server.metadata.TabletServerResource;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TabletServerResourceProbe}. */
+class TabletServerResourceProbeTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void testProbeFromExplicitConfig() {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES, 8.0);
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE, new MemorySize(1024));
+
+ TabletServerResource resource = new TabletServerResourceProbe(conf, tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(8.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(1024);
+ }
+
+ @Test
+ void testProbeFromCgroupV2Files() throws Exception {
+ write(tempDir.resolve("cpu.max"), "200000 100000");
+ write(tempDir.resolve("memory.max"), "1024");
+
+ TabletServerResource resource =
+ new TabletServerResourceProbe(new Configuration(), tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(2.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(1024);
+ }
+
+ @Test
+ void testProbeFromCgroupV1Files() throws Exception {
+ Path cpuDir = Files.createDirectory(tempDir.resolve("cpu"));
+ Path memoryDir = Files.createDirectory(tempDir.resolve("memory"));
+ write(cpuDir.resolve("cpu.cfs_quota_us"), "250000");
+ write(cpuDir.resolve("cpu.cfs_period_us"), "100000");
+ write(memoryDir.resolve("memory.limit_in_bytes"), "2048");
+
+ TabletServerResource resource =
+ new TabletServerResourceProbe(new Configuration(), tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(2.5);
+ assertThat(resource.getMemoryBytes()).isEqualTo(2048);
+ }
+
+ @Test
+ void testProbeFallsBackToCgroupV1WhenCgroupV2IsUnlimited() throws Exception {
+ write(tempDir.resolve("cpu.max"), "max 100000");
+ write(tempDir.resolve("memory.max"), "max");
+ Path cpuDir = Files.createDirectory(tempDir.resolve("cpu"));
+ Path memoryDir = Files.createDirectory(tempDir.resolve("memory"));
+ write(cpuDir.resolve("cpu.cfs_quota_us"), "300000");
+ write(cpuDir.resolve("cpu.cfs_period_us"), "100000");
+ write(memoryDir.resolve("memory.limit_in_bytes"), "4096");
+
+ TabletServerResource resource =
+ new TabletServerResourceProbe(new Configuration(), tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(3.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(4096);
+ }
+
+ @Test
+ void testProbeFallsBackToCgroupWhenExplicitConfigIsNonPositive() throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES, 0.0);
+ conf.set(ConfigOptions.TABLET_SERVER_ADVERTISED_RESOURCE_MEMORY_SIZE, new MemorySize(0));
+ write(tempDir.resolve("cpu.max"), "400000 100000");
+ write(tempDir.resolve("memory.max"), "8192");
+
+ TabletServerResource resource = new TabletServerResourceProbe(conf, tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isEqualTo(4.0);
+ assertThat(resource.getMemoryBytes()).isEqualTo(8192);
+ }
+
+ @Test
+ void testProbeFallsBackToJavaRuntimeAndOperatingSystem() {
+ TabletServerResource resource =
+ new TabletServerResourceProbe(new Configuration(), tempDir).probe();
+
+ assertThat(resource.getCpuCores()).isNotNull().isGreaterThan(0.0);
+
+ java.lang.management.OperatingSystemMXBean operatingSystemMXBean =
+ java.lang.management.ManagementFactory.getOperatingSystemMXBean();
+ if (operatingSystemMXBean instanceof com.sun.management.OperatingSystemMXBean) {
+ long expectedMemoryBytes =
+ ((com.sun.management.OperatingSystemMXBean) operatingSystemMXBean)
+ .getTotalPhysicalMemorySize();
+ if (expectedMemoryBytes > 0) {
+ assertThat(resource.getMemoryBytes()).isEqualTo(expectedMemoryBytes);
+ } else {
+ assertThat(resource.getMemoryBytes()).isNull();
+ }
+ } else {
+ assertThat(resource.getMemoryBytes()).isNull();
+ }
+ }
+
+ private static void write(Path path, String value) throws Exception {
+ Files.write(path, Collections.singletonList(value), StandardCharsets.UTF_8);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerdeTest.java
index 575e5dba2..01359be76 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerdeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/TabletServerRegistrationJsonSerdeTest.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.zk.data;
import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.server.metadata.TabletServerResource;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.fluss.utils.json.JsonSerdeTestBase;
@@ -49,16 +50,24 @@ public class TabletServerRegistrationJsonSerdeTest
Endpoint.fromListenersString(
"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346"),
10000);
+ TabletServerRegistration tabletServerRegistration3 =
+ new TabletServerRegistration(
+ "cn-hangzhou-server10",
+ Endpoint.fromListenersString(
+ "CLIENT://localhost:2345,FLUSS://127.0.0.1:2346"),
+ 10000,
+ new TabletServerResource(8.0, 1024L));
return new TabletServerRegistration[] {
- tabletServerRegistration1, tabletServerRegistration2
+ tabletServerRegistration1, tabletServerRegistration2, tabletServerRegistration3
};
}
@Override
protected String[] expectedJsons() {
return new String[] {
- "{\"version\":3,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\",\"register_timestamp\":10000}",
- "{\"version\":3,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\",\"register_timestamp\":10000,\"rack\":\"cn-hangzhou-server10\"}"
+ "{\"version\":4,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\",\"register_timestamp\":10000}",
+ "{\"version\":4,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\",\"register_timestamp\":10000,\"rack\":\"cn-hangzhou-server10\"}",
+ "{\"version\":4,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\",\"register_timestamp\":10000,\"rack\":\"cn-hangzhou-server10\",\"cpu_cores\":8.0,\"memory_bytes\":1024}"
};
}
@@ -94,5 +103,22 @@ public class TabletServerRegistrationJsonSerdeTest
"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346"),
10000);
assertEquals(tabletServerRegistration, expectedTabletServerRegistration);
+
+ // compatibility with version 3
+ JsonNode jsonInVersion3 =
+ new ObjectMapper()
+ .readTree(
+ ("{\"version\":3,\"listeners\":\"CLIENT://localhost:2345,FLUSS://127.0.0.1:2346\","
+ + "\"register_timestamp\":10000,\"rack\":\"cn-hangzhou-server10\"}")
+ .getBytes(StandardCharsets.UTF_8));
+ tabletServerRegistration =
+ TabletServerRegistrationJsonSerde.INSTANCE.deserialize(jsonInVersion3);
+ expectedTabletServerRegistration =
+ new TabletServerRegistration(
+ "cn-hangzhou-server10",
+ Endpoint.fromListenersString(
+ "CLIENT://localhost:2345,FLUSS://127.0.0.1:2346"),
+ 10000);
+ assertEquals(tabletServerRegistration, expectedTabletServerRegistration);
}
}