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 e5e6bb49c [server] Add configurable disk write-limit recover ratio
(#3612)
e5e6bb49c is described below
commit e5e6bb49cee5d2c0a8e167f6588c206a227e0f87
Author: yunhong <[email protected]>
AuthorDate: Mon Jul 13 20:37:42 2026 +0800
[server] Add configurable disk write-limit recover ratio (#3612)
---
.../fluss/client/admin/FlussAdminITCase.java | 152 +++++++-------
.../org/apache/fluss/config/ConfigOptions.java | 19 +-
.../apache/fluss/server/DynamicServerConfig.java | 2 +
.../server/coordinator/CoordinatorServer.java | 4 +-
.../fluss/server/storage/DiskUsageMonitor.java | 74 ++++---
.../storage/DiskWriteLimitConfigValidator.java | 65 ++++++
.../storage/DiskWriteLimitRatioValidator.java | 60 ------
.../fluss/server/storage/LocalDiskManager.java | 94 ++++++---
.../apache/fluss/server/tablet/TabletServer.java | 2 +-
.../fluss/server/DynamicConfigChangeTest.java | 223 ++++++++++++---------
.../fluss/server/replica/ReplicaManagerTest.java | 2 +-
.../fluss/server/storage/DiskUsageMonitorTest.java | 116 +++++++----
.../fluss/server/storage/LocalDiskManagerTest.java | 34 ++++
website/docs/maintenance/configuration.md | 33 +--
14 files changed, 541 insertions(+), 339 deletions(-)
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index cda9fa334..ea60dc3a0 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -1652,76 +1652,92 @@ class FlussAdminITCase extends ClientToServerITCaseBase
{
}
@Test
- void testDynamicDiskWriteLimitRatio() throws Exception {
- // Valid value should succeed
- admin.alterClusterConfigs(
- Collections.singletonList(
- new AlterConfig(
-
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
- "0.15",
- AlterConfigOpType.SET)))
- .get();
- assertConfigEntry(
- ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
- "0.15",
- ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
-
- // Invalid value: 0.0 (must be > 0.1)
- assertThatThrownBy(
- () ->
- admin.alterClusterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
- .key(),
- "0.0",
-
AlterConfigOpType.SET)))
- .get())
- .cause()
- .isInstanceOf(ConfigException.class)
- .hasMessageContaining("must be within (0.1, 1.0]");
-
- // Invalid value: 0.1 (boundary, must be > 0.1)
- assertThatThrownBy(
- () ->
- admin.alterClusterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
- .key(),
- "0.1",
-
AlterConfigOpType.SET)))
- .get())
- .cause()
- .isInstanceOf(ConfigException.class)
- .hasMessageContaining("must be within (0.1, 1.0]");
+ void testDynamicDiskWriteLimitRatios() throws Exception {
+ try {
+ // Both ratios can be lowered atomically when their relationship
remains valid.
+ admin.alterClusterConfigs(
+ Arrays.asList(
+ new AlterConfig(
+
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ "0.95",
+ AlterConfigOpType.SET),
+ new AlterConfig(
+
ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO
+ .key(),
+ "0.90",
+ AlterConfigOpType.SET)))
+ .get();
+ assertConfigEntry(
+ ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ "0.95",
+ ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
+ assertConfigEntry(
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
+ "0.90",
+ ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
+
+ // Invalid value: 0.0 (must be greater than the recover ratio).
+ assertThatThrownBy(
+ () ->
+ admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions
+
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
+
.key(),
+ "0.0",
+
AlterConfigOpType.SET)))
+ .get())
+ .cause()
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("Invalid disk write-limit
configuration");
+ // Invalid value: 0.90 (must be strictly greater than the recover
ratio).
+ assertThatThrownBy(
+ () ->
+ admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions
+
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
+
.key(),
+ "0.90",
+
AlterConfigOpType.SET)))
+ .get())
+ .cause()
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("Invalid disk write-limit
configuration");
- // Invalid value: 1.5 (must be <= 1.0)
- assertThatThrownBy(
- () ->
- admin.alterClusterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
- .key(),
- "1.5",
-
AlterConfigOpType.SET)))
- .get())
- .cause()
- .isInstanceOf(ConfigException.class)
- .hasMessageContaining("must be within (0.1, 1.0]");
+ // Invalid value: 1.5 (must be <= 1.0)
+ assertThatThrownBy(
+ () ->
+ admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions
+
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
+
.key(),
+ "1.5",
+
AlterConfigOpType.SET)))
+ .get())
+ .cause()
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("Invalid disk write-limit
configuration");
- // Reset should succeed (restores default)
- admin.alterClusterConfigs(
- Collections.singletonList(
- new AlterConfig(
-
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
- null,
- AlterConfigOpType.DELETE)))
- .get();
+ } finally {
+ // Reset both ratios in one request so the default relationship
remains valid.
+ admin.alterClusterConfigs(
+ Arrays.asList(
+ new AlterConfig(
+
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ null,
+ AlterConfigOpType.DELETE),
+ new AlterConfig(
+
ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO
+ .key(),
+ null,
+ AlterConfigOpType.DELETE)))
+ .get();
+ }
}
@Test
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 73eccf1b7..1914df7e1 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
@@ -385,10 +385,23 @@ public class ConfigOptions {
.doubleType()
.defaultValue(0.85)
.withDescription(
- "Reject writes when the tablet server data disk
usage exceeds this ratio. "
- + "Writes resume after the usage drops
below (ratio - 0.10). "
+ "Reject writes when the tablet server data disk
usage reaches this ratio. "
+ + "Writes resume when the usage reaches or
drops below "
+ + "server.data-disk.write-recover-ratio. "
+ "Set to 1.0 to disable the disk-usage
protection entirely. "
- + "The valid range is (0.1, 1.0].");
+ + "The valid range is "
+ + "(server.data-disk.write-recover-ratio,
1.0]. When lowering "
+ + "both ratios dynamically, update them in
the same request or "
+ + "lower
server.data-disk.write-recover-ratio first.");
+
+ public static final ConfigOption<Double>
SERVER_DATA_DISK_WRITE_RECOVER_RATIO =
+ key("server.data-disk.write-recover-ratio")
+ .doubleType()
+ .defaultValue(0.80)
+ .withDescription(
+ "Resume writes when the tablet server data disk
usage reaches or "
+ + "drops below this ratio. The valid range
is "
+ + "(0.0,
server.data-disk.write-limit-ratio).");
public static final ConfigOption<Duration> SERVER_DATA_DISK_CHECK_INTERVAL
=
key("server.data-disk.check-interval")
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 710c6b675..269e0c483 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
@@ -53,6 +53,7 @@ import static
org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS;
import static
org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO;
+import static
org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO;
import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_CREDENTIALS;
import static
org.apache.fluss.config.ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
@@ -75,6 +76,7 @@ class DynamicServerConfig {
LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
KV_SNAPSHOT_INTERVAL.key(),
+ SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
// Config options for remote.data.dirs
REMOTE_DATA_DIRS.key(),
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 d4fbcbb97..ba9287b27 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
@@ -40,7 +40,7 @@ import org.apache.fluss.server.metadata.ServerMetadataCache;
import org.apache.fluss.server.metrics.ServerMetricUtils;
import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
import org.apache.fluss.server.metrics.group.LakeTieringMetricGroup;
-import org.apache.fluss.server.storage.DiskWriteLimitRatioValidator;
+import org.apache.fluss.server.storage.DiskWriteLimitConfigValidator;
import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperUtils;
@@ -299,7 +299,7 @@ public class CoordinatorServer extends ServerBase {
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.register(remoteDirDynamicLoader);
// Register stateless validators for coordinator-side upfront
validation
- dynamicConfigManager.registerValidator(new
DiskWriteLimitRatioValidator());
+ dynamicConfigManager.register(new DiskWriteLimitConfigValidator());
rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register);
dynamicConfigManager.startup();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageMonitor.java
b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageMonitor.java
index 8a27c1ef3..16f7d90a8 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageMonitor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskUsageMonitor.java
@@ -30,38 +30,36 @@ import static
org.apache.fluss.utils.Preconditions.checkNotNull;
/**
* Periodically samples the local data disk usage ratio and toggles the tablet
server write-lock
- * state with a fixed 10% hysteresis: writes are locked when the usage reaches
the configured
- * write-limit ratio and resume only after the usage drops below {@code (limit
- 0.10)}. The monitor
- * is single-state and intended to be driven by a scheduler thread; it never
blocks.
+ * state with a configurable hysteresis: writes are locked when the usage
reaches the configured
+ * write-limit ratio and resume only after the usage reaches or drops below
the configured
+ * write-recover ratio. The monitor is single-state and intended to be driven
by a scheduler thread;
+ * it never blocks.
*/
@Internal
public final class DiskUsageMonitor {
- /** Fixed hysteresis between the lock and unlock thresholds. */
- public static final double RECOVER_GAP = 0.10;
-
private static final Logger LOG =
LoggerFactory.getLogger(DiskUsageMonitor.class);
private final int serverId;
private final DiskUsageCollector collector;
private volatile double writeLimitRatio;
- private volatile double recoverThreshold;
+ private volatile double writeRecoverRatio;
private final Listener listener;
private volatile boolean locked;
private volatile double lastUsageRatio;
public DiskUsageMonitor(
- int serverId, DiskUsageCollector collector, double
writeLimitRatio, Listener listener) {
- checkArgument(
- writeLimitRatio > 0.0 && writeLimitRatio <= 1.0,
- "%s must be within (0.0, 1.0], but was %s",
- "server.data-disk.write-limit-ratio",
- writeLimitRatio);
+ int serverId,
+ DiskUsageCollector collector,
+ double writeLimitRatio,
+ double writeRecoverRatio,
+ Listener listener) {
+ checkValidWriteLimitConfig(writeLimitRatio, writeRecoverRatio);
this.serverId = serverId;
this.collector = checkNotNull(collector, "collector");
this.writeLimitRatio = writeLimitRatio;
- this.recoverThreshold = Math.max(0.0, writeLimitRatio - RECOVER_GAP);
+ this.writeRecoverRatio = writeRecoverRatio;
this.listener = checkNotNull(listener, "listener");
}
@@ -85,7 +83,7 @@ public final class DiskUsageMonitor {
serverId,
String.format("%.4f", usage * 100),
String.format("%.2f", writeLimitRatio * 100),
- String.format("%.2f", recoverThreshold * 100),
+ String.format("%.2f", writeRecoverRatio * 100),
locked);
}
update(usage);
@@ -108,19 +106,19 @@ public final class DiskUsageMonitor {
locked = true;
LOG.warn(
"TabletServer {} disk usage reached {}% (limit {}%);
rejecting writes "
- + "until usage drops below {}%.",
+ + "until usage reaches or drops below {}%.",
serverId,
String.format("%.2f", usage * 100),
String.format("%.2f", writeLimitRatio * 100),
- String.format("%.2f", recoverThreshold * 100));
- } else if (wasLocked && usage <= recoverThreshold) {
+ String.format("%.2f", writeRecoverRatio * 100));
+ } else if (wasLocked && usage <= writeRecoverRatio) {
locked = false;
LOG.info(
"TabletServer {} disk usage dropped to {}% (recover
threshold {}%); "
+ "resuming writes.",
serverId,
String.format("%.2f", usage * 100),
- String.format("%.2f", recoverThreshold * 100));
+ String.format("%.2f", writeRecoverRatio * 100));
}
listener.onSample(lastUsageRatio, locked);
}
@@ -137,23 +135,39 @@ public final class DiskUsageMonitor {
return writeLimitRatio;
}
- public double getRecoverThreshold() {
- return recoverThreshold;
+ public double getWriteRecoverRatio() {
+ return writeRecoverRatio;
}
/**
- * Dynamically updates the write-limit ratio and the derived recover
threshold. The new ratio
- * takes effect on the next {@link #runOnce()} invocation or {@link
#update(double)} call.
+ * Dynamically updates the write-limit and write-recover ratios. The new
values take effect on
+ * the next {@link #runOnce()} invocation or {@link #update(double)} call.
*
- * @param newRatio the new write-limit ratio, must be within (0.0, 1.0]
+ * @param newRatio the new write-limit ratio, must be within
(newRecoverRatio, 1.0]
+ * @param newRecoverRatio the new write-recover ratio, must be within
(0.0, newRatio)
*/
- public void updateWriteLimitRatio(double newRatio) {
- checkArgument(
- newRatio > 0.0 && newRatio <= 1.0,
- "server.data-disk.write-limit-ratio must be within (0.0, 1.0],
but was %s",
- newRatio);
+ public void updateWriteLimitConfig(double newRatio, double
newRecoverRatio) {
+ checkValidWriteLimitConfig(newRatio, newRecoverRatio);
this.writeLimitRatio = newRatio;
- this.recoverThreshold = Math.max(0.0, newRatio - RECOVER_GAP);
+ this.writeRecoverRatio = newRecoverRatio;
+ }
+
+ /**
+ * Dynamically updates the write-limit ratio while keeping the current
write-recover ratio.
+ *
+ * @param newRatio the new write-limit ratio, must be greater than the
current write-recover
+ * ratio and no greater than 1.0
+ */
+ public void updateWriteLimitRatio(double newRatio) {
+ updateWriteLimitConfig(newRatio, writeRecoverRatio);
+ }
+
+ private static void checkValidWriteLimitConfig(
+ double writeLimitRatio, double writeRecoverRatio) {
+ String validationError =
+ DiskWriteLimitConfigValidator.getValidationError(
+ writeLimitRatio, writeRecoverRatio);
+ checkArgument(validationError == null, validationError);
}
/** Receives every sample for downstream state synchronization (e.g.
metrics gauges). */
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitConfigValidator.java
b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitConfigValidator.java
new file mode 100644
index 000000000..fcc103fdb
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitConfigValidator.java
@@ -0,0 +1,65 @@
+/*
+ * 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.storage;
+
+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;
+
+/**
+ * Validator for the data disk write-limit configuration.
+ *
+ * <p>This validator is registered on the CoordinatorServer so invalid dynamic
updates are rejected
+ * before they are persisted to ZooKeeper. The TabletServer performs the same
validation through
+ * {@link LocalDiskManager}.
+ */
+public class DiskWriteLimitConfigValidator implements ServerReconfigurable {
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ String validationError =
+ getValidationError(
+
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO),
+
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO));
+ if (validationError != null) {
+ throw new ConfigException(validationError);
+ }
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) {}
+
+ static String getValidationError(double writeLimitRatio, double
writeRecoverRatio) {
+ if (writeRecoverRatio > 0.0
+ && writeRecoverRatio < writeLimitRatio
+ && writeLimitRatio <= 1.0) {
+ return null;
+ }
+ return String.format(
+ "Invalid disk write-limit configuration: %s must be within
(0.0, %s), and %s "
+ + "must be no greater than 1.0; %s=%s, %s=%s",
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
+ ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ writeLimitRatio,
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
+ writeRecoverRatio);
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitRatioValidator.java
b/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitRatioValidator.java
deleted file mode 100644
index 14fbbaa2b..000000000
---
a/fluss-server/src/main/java/org/apache/fluss/server/storage/DiskWriteLimitRatioValidator.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.storage;
-
-import org.apache.fluss.config.ConfigOptions;
-import org.apache.fluss.config.cluster.ConfigValidator;
-import org.apache.fluss.exception.ConfigException;
-
-import javax.annotation.Nullable;
-
-/**
- * Stateless validator for {@link
ConfigOptions#SERVER_DATA_DISK_WRITE_LIMIT_RATIO}.
- *
- * <p>This validator ensures the disk write-limit ratio stays within the valid
range (0.1, 1.0]. The
- * lower bound of 0.1 (exclusive) guarantees the recover threshold (ratio -
0.10) remains positive,
- * ensuring the system can always recover from a write-locked state. It is
registered on the
- * CoordinatorServer so that invalid values are rejected upfront during {@code
AlterConfigs}, before
- * being persisted to ZooKeeper.
- *
- * <p>The same range check is also performed on the TabletServer side by
{@link LocalDiskManager}
- * (via {@link
org.apache.fluss.config.cluster.ServerReconfigurable#validate}), providing
- * defense-in-depth.
- */
-public class DiskWriteLimitRatioValidator implements ConfigValidator<Double> {
-
- @Override
- public String configKey() {
- return ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key();
- }
-
- @Override
- public void validate(@Nullable Double oldValue, @Nullable Double newValue)
- throws ConfigException {
- if (newValue == null) {
- // Deletion (reset to default) is always valid.
- return;
- }
- if (newValue <= 0.1 || newValue > 1.0) {
- throw new ConfigException(
- String.format(
- "Invalid %s: must be within (0.1, 1.0], but was
%s",
-
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), newValue));
- }
- }
-}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
index a1afe73bc..b12e897ab 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
@@ -102,12 +102,17 @@ public final class LocalDiskManager implements Closeable,
ServerReconfigurable {
/**
* Configured high-water-mark ratio: writes are rejected once any single
backing {@link
* java.nio.file.FileStore}'s usage ratio reaches this value. Writes
resume after the usage
- * drops below {@code diskWriteLimitRatio - 0.10} (the recover gap is
owned by {@link
- * DiskUsageMonitor}). This field is volatile because it can be changed at
runtime via dynamic
- * reconfiguration.
+ * reaches or drops below {@link #diskWriteRecoverRatio}. This field is
volatile because it can
+ * be changed at runtime via dynamic reconfiguration.
*/
private volatile double diskWriteLimitRatio;
+ /**
+ * Configured low-water-mark ratio at which writes resume. This field is
volatile because it can
+ * be changed at runtime via dynamic reconfiguration.
+ */
+ private volatile double diskWriteRecoverRatio;
+
/**
* Whether the local tablet server is currently rejecting writes because
the data disk usage has
* reached {@link #diskWriteLimitRatio}. Updated by the {@link
DiskUsageMonitor} scheduler
@@ -124,23 +129,13 @@ public final class LocalDiskManager implements Closeable,
ServerReconfigurable {
private LocalDiskManager(Configuration conf) throws IOException {
this.serverId = conf.getInt(ConfigOptions.TABLET_SERVER_ID);
- List<File> configuredDataDirs = resolveConfiguredDataDirs(conf);
- List<File> initializedDataDirs;
- try {
- initializedDataDirs = initializeDataDirs(configuredDataDirs);
- } catch (Throwable throwable) {
- close();
- throw throwable;
- }
- this.dataDirs = Collections.unmodifiableList(initializedDataDirs);
-
this.diskWriteLimitRatio =
conf.get(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO);
- if (diskWriteLimitRatio <= 0.0 || diskWriteLimitRatio > 1.0) {
- throw new IllegalConfigurationException(
- String.format(
- "%s must be within (0.0, 1.0], but was %s",
-
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
- diskWriteLimitRatio));
+ this.diskWriteRecoverRatio =
conf.get(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO);
+ String diskWriteLimitValidationError =
+ DiskWriteLimitConfigValidator.getValidationError(
+ diskWriteLimitRatio, diskWriteRecoverRatio);
+ if (diskWriteLimitValidationError != null) {
+ throw new
IllegalConfigurationException(diskWriteLimitValidationError);
}
this.diskCheckIntervalMs =
conf.get(ConfigOptions.SERVER_DATA_DISK_CHECK_INTERVAL).toMillis();
@@ -151,11 +146,22 @@ public final class LocalDiskManager implements Closeable,
ServerReconfigurable {
ConfigOptions.SERVER_DATA_DISK_CHECK_INTERVAL.key(),
diskCheckIntervalMs));
}
+
+ List<File> configuredDataDirs = resolveConfiguredDataDirs(conf);
+ List<File> initializedDataDirs;
+ try {
+ initializedDataDirs = initializeDataDirs(configuredDataDirs);
+ } catch (Throwable throwable) {
+ close();
+ throw throwable;
+ }
+ this.dataDirs = Collections.unmodifiableList(initializedDataDirs);
this.diskUsageMonitor =
new DiskUsageMonitor(
serverId,
new DiskUsageCollector(this.dataDirs),
diskWriteLimitRatio,
+ diskWriteRecoverRatio,
(usage, locked) -> {
this.lastDiskUsageRatio = usage;
this.diskWriteLocked = locked;
@@ -497,6 +503,13 @@ public final class LocalDiskManager implements Closeable,
ServerReconfigurable {
return diskWriteLimitRatio;
}
+ /**
+ * @return the configured low-water-mark ratio at which writes resume.
+ */
+ public double getDiskWriteRecoverRatio() {
+ return diskWriteRecoverRatio;
+ }
+
/**
* Throws {@link DiskWriteLockedException} when the local data disk usage
has crossed the
* configured write-limit ratio. Only client-driven writes ({@code
appendLog} / {@code putKv})
@@ -515,41 +528,56 @@ public final class LocalDiskManager implements Closeable,
ServerReconfigurable {
}
// ------------------------------------------------------------------------
- // ServerReconfigurable: dynamic write-limit-ratio update
+ // ServerReconfigurable: dynamic disk write-limit update
// ------------------------------------------------------------------------
@Override
public void validate(Configuration newConfig) throws ConfigException {
double newRatio =
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO);
- if (newRatio <= 0.1 || newRatio > 1.0) {
- throw new ConfigException(
- String.format(
- "Invalid %s: must be within (0.1, 1.0], but was
%s",
-
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), newRatio));
- }
+ double newRecoverRatio =
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO);
+ validateDiskWriteLimitConfig(newRatio, newRecoverRatio);
}
@Override
- public void reconfigure(Configuration newConfig) {
+ public void reconfigure(Configuration newConfig) throws ConfigException {
double newRatio =
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO);
- if (Double.compare(newRatio, diskWriteLimitRatio) == 0) {
+ double newRecoverRatio =
newConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO);
+ validateDiskWriteLimitConfig(newRatio, newRecoverRatio);
+ if (Double.compare(newRatio, diskWriteLimitRatio) == 0
+ && Double.compare(newRecoverRatio, diskWriteRecoverRatio) ==
0) {
LOG.debug(
- "{} unchanged: {}",
+ "{}/{} unchanged: {}/{}",
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
- newRatio);
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
+ newRatio,
+ newRecoverRatio);
return;
}
double oldRatio = diskWriteLimitRatio;
+ double oldRecoverRatio = diskWriteRecoverRatio;
diskWriteLimitRatio = newRatio;
- diskUsageMonitor.updateWriteLimitRatio(newRatio);
+ diskWriteRecoverRatio = newRecoverRatio;
+ diskUsageMonitor.updateWriteLimitConfig(newRatio, newRecoverRatio);
// Trigger an immediate check so the new threshold takes effect
without waiting
// for the next scheduled tick.
diskUsageMonitor.runOnce();
LOG.info(
- "{} reconfigured: {} -> {} (immediate check triggered)",
+ "{}/{} reconfigured: {}/{} -> {}/{} (immediate check
triggered)",
ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
oldRatio,
- newRatio);
+ oldRecoverRatio,
+ newRatio,
+ newRecoverRatio);
+ }
+
+ private static void validateDiskWriteLimitConfig(double newRatio, double
newRecoverRatio)
+ throws ConfigException {
+ String validationError =
+ DiskWriteLimitConfigValidator.getValidationError(newRatio,
newRecoverRatio);
+ if (validationError != null) {
+ throw new ConfigException(validationError);
+ }
}
/**
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 adf9a59bd..b73390fdc 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
@@ -323,7 +323,7 @@ public class TabletServer extends ServerBase {
dynamicConfigManager.register(replicaManager.getKvSnapshotContext());
// Register replicaManager to dynamicConfigManager for dynamic
config
dynamicConfigManager.register(replicaManager);
- // Register localDiskManager for dynamic
server.data-disk.write-limit-ratio
+ // Register localDiskManager for dynamic disk write-limit and
recover ratios.
dynamicConfigManager.register(localDiskManager);
rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register);
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 839c8c69a..a311a2986 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
@@ -28,6 +28,7 @@ import
org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
import org.apache.fluss.server.coordinator.remote.RoundRobinRemoteDirSelector;
import
org.apache.fluss.server.coordinator.remote.WeightedRoundRobinRemoteDirSelector;
+import org.apache.fluss.server.storage.DiskWriteLimitConfigValidator;
import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -58,9 +59,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;
-import static org.assertj.core.api.Assertions.within;
/** Test for {@link DynamicConfigManager}. */
public class DynamicConfigChangeTest {
@@ -207,15 +206,10 @@ public class DynamicConfigChangeTest {
// Setting `datalake.paimon.*` without setting `datalake.format`
should pass because
// prefix validation is skipped.
- assertThatCode(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
-
"datalake.iceberg.type",
- "rest",
-
AlterConfigOpType.SET))))
- .doesNotThrowAnyException();
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ "datalake.iceberg.type", "rest",
AlterConfigOpType.SET)));
assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
.isNull();
@@ -351,17 +345,12 @@ public class DynamicConfigChangeTest {
dynamicConfigManager.startup();
// Adjust rate limiter value - should succeed
- assertThatCode(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC
- .key(),
- "200MB",
-
AlterConfigOpType.SET))))
- .doesNotThrowAnyException();
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
+ "200MB",
+ AlterConfigOpType.SET)));
// Verify config was persisted to ZK
Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
@@ -448,15 +437,12 @@ public class DynamicConfigChangeTest {
dynamicConfigManager.startup();
// Change snapshot interval to 5 minutes - should succeed
- assertThatCode(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
-
ConfigOptions.KV_SNAPSHOT_INTERVAL.key(),
- "5min",
-
AlterConfigOpType.SET))))
- .doesNotThrowAnyException();
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.KV_SNAPSHOT_INTERVAL.key(),
+ "5min",
+ AlterConfigOpType.SET)));
// Verify config was persisted to ZK
Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
@@ -569,17 +555,12 @@ public class DynamicConfigChangeTest {
dynamicConfigManager.startup();
// Change min-in-sync-replicas to 2 - should succeed
- assertThatCode(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER
- .key(),
- "2",
-
AlterConfigOpType.SET))))
- .doesNotThrowAnyException();
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
+ "2",
+ AlterConfigOpType.SET)));
// Verify config was persisted to ZK
Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
@@ -614,7 +595,50 @@ public class DynamicConfigChangeTest {
}
@Test
- void testDynamicDiskWriteLimitRatioChange(@TempDir File tempDir) throws
Exception {
+ void testCoordinatorValidatesDiskWriteLimitConfigAgainstDefaults() throws
Exception {
+ DynamicConfigManager dynamicConfigManager =
createDiskWriteLimitDynamicConfigManager();
+
+ alterDiskWriteRecoverRatio(dynamicConfigManager, "0.70");
+
+ assertThatThrownBy(() ->
alterDiskWriteRecoverRatio(dynamicConfigManager, "0.0"))
+ .isInstanceOf(ConfigException.class);
+
+ assertThatThrownBy(() ->
alterDiskWriteLimitRatio(dynamicConfigManager, "0.70"))
+ .isInstanceOf(ConfigException.class);
+
+ alterDiskWriteLimitRatio(dynamicConfigManager, "1.0");
+
+ assertThatThrownBy(() ->
alterDiskWriteLimitRatio(dynamicConfigManager, "1.1"))
+ .isInstanceOf(ConfigException.class);
+
+ assertThat(zookeeperClient.fetchEntityConfig())
+
.containsEntry(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), "1.0")
+
.containsEntry(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
"0.70");
+ }
+
+ @Test
+ void testCoordinatorValidatesRecoverRatioWhenLoweringDiskWriteLimitRatio()
throws Exception {
+ DynamicConfigManager dynamicConfigManager =
createDiskWriteLimitDynamicConfigManager();
+
+ assertThatThrownBy(() ->
alterDiskWriteLimitRatio(dynamicConfigManager, "0.70"))
+ .isInstanceOf(ConfigException.class);
+
+ assertThat(zookeeperClient.fetchEntityConfig())
+
.doesNotContainKey(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key());
+
+ alterDiskWriteRecoverRatio(dynamicConfigManager, "0.60");
+ alterDiskWriteLimitRatio(dynamicConfigManager, "0.70");
+
+ assertThatThrownBy(() ->
alterDiskWriteRecoverRatio(dynamicConfigManager, "0.70"))
+ .isInstanceOf(ConfigException.class);
+
+ assertThat(zookeeperClient.fetchEntityConfig())
+
.containsEntry(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), "0.70")
+
.containsEntry(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
"0.60");
+ }
+
+ @Test
+ void testDynamicDiskWriteLimitConfigChange(@TempDir File tempDir) throws
Exception {
File dataDir = new File(tempDir, "data-0");
assertThat(dataDir.mkdirs()).isTrue();
@@ -633,70 +657,89 @@ public class DynamicConfigChangeTest {
// Verify initial state
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.85);
+
assertThat(localDiskManager.getDiskWriteRecoverRatio()).isEqualTo(0.80);
assertThat(localDiskManager.getDiskUsageMonitor().getWriteLimitRatio()).isEqualTo(0.85);
-
assertThat(localDiskManager.getDiskUsageMonitor().getRecoverThreshold())
- .isEqualTo(0.75);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteRecoverRatio())
+ .isEqualTo(0.80);
- // Lower the limit to 0.70 via dynamic config
- assertThatCode(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
- .key(),
- "0.70",
-
AlterConfigOpType.SET))))
- .doesNotThrowAnyException();
+ // Lower the recover ratio before lowering the limit.
+ alterDiskWriteRecoverRatio(dynamicConfigManager, "0.60");
+ alterDiskWriteLimitRatio(dynamicConfigManager, "0.70");
- // Verify the new ratio took effect immediately (reconfigure
triggers runOnce)
+ // Verify both ratios took effect immediately (reconfigure
triggers runOnce).
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.70);
+
assertThat(localDiskManager.getDiskWriteRecoverRatio()).isEqualTo(0.60);
assertThat(localDiskManager.getDiskUsageMonitor().getWriteLimitRatio()).isEqualTo(0.70);
-
assertThat(localDiskManager.getDiskUsageMonitor().getRecoverThreshold())
- .isCloseTo(0.60, within(1e-9));
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteRecoverRatio())
+ .isEqualTo(0.60);
+
+ alterDiskWriteRecoverRatio(dynamicConfigManager, "0.65");
+
+
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.70);
+
assertThat(localDiskManager.getDiskWriteRecoverRatio()).isEqualTo(0.65);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteLimitRatio()).isEqualTo(0.70);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteRecoverRatio())
+ .isEqualTo(0.65);
// Verify config was persisted to ZK
Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
assertThat(zkConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key()))
.isEqualTo("0.70");
- }
- }
+
assertThat(zkConfig.get(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key()))
+ .isEqualTo("0.65");
- @Test
- void testPreventInvalidDiskWriteLimitRatio(@TempDir File tempDir) throws
Exception {
- File dataDir = new File(tempDir, "data-0");
- assertThat(dataDir.mkdirs()).isTrue();
+ assertThatThrownBy(() ->
alterDiskWriteLimitRatio(dynamicConfigManager, "0.0"))
+ .isInstanceOf(ConfigException.class);
- Configuration configuration = new Configuration();
- configuration.setInt(ConfigOptions.TABLET_SERVER_ID, 0);
- configuration.setString(ConfigOptions.DATA_DIR,
dataDir.getAbsolutePath());
- configuration.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO,
0.85);
+ assertThatThrownBy(() ->
alterDiskWriteRecoverRatio(dynamicConfigManager, "0.0"))
+ .isInstanceOf(ConfigException.class);
+
+ assertThatThrownBy(() ->
alterDiskWriteRecoverRatio(dynamicConfigManager, "0.70"))
+ .isInstanceOf(ConfigException.class);
+ Configuration invalidReconfigure = new
Configuration(configuration);
+
invalidReconfigure.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO, 0.65);
+
invalidReconfigure.set(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO,
0.65);
+ assertThatThrownBy(() ->
localDiskManager.reconfigure(invalidReconfigure))
+ .isInstanceOf(ConfigException.class);
+
+
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.70);
+
assertThat(localDiskManager.getDiskWriteRecoverRatio()).isEqualTo(0.65);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteLimitRatio()).isEqualTo(0.70);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteRecoverRatio())
+ .isEqualTo(0.65);
+ }
+ }
+
+ private static DynamicConfigManager
createDiskWriteLimitDynamicConfigManager()
+ throws Exception {
DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ new DynamicConfigManager(zookeeperClient, new Configuration(),
true);
+ dynamicConfigManager.register(new DiskWriteLimitConfigValidator());
+ dynamicConfigManager.startup();
+ return dynamicConfigManager;
+ }
- try (LocalDiskManager localDiskManager =
LocalDiskManager.create(configuration)) {
- dynamicConfigManager.register(localDiskManager);
- dynamicConfigManager.startup();
+ private static void alterDiskWriteLimitRatio(
+ DynamicConfigManager dynamicConfigManager, String value) throws
Exception {
+ alterConfig(
+ dynamicConfigManager,
+ ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+ value);
+ }
- // Try to set to invalid value 0.0 - should be rejected
- assertThatThrownBy(
- () ->
- dynamicConfigManager.alterConfigs(
- Collections.singletonList(
- new AlterConfig(
- ConfigOptions
-
.SERVER_DATA_DISK_WRITE_LIMIT_RATIO
- .key(),
- "0.0",
-
AlterConfigOpType.SET))))
- .isInstanceOf(ConfigException.class)
- .hasMessageContaining("must be within (0.1, 1.0]");
+ private static void alterDiskWriteRecoverRatio(
+ DynamicConfigManager dynamicConfigManager, String value) throws
Exception {
+ alterConfig(
+ dynamicConfigManager,
+ ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(),
+ value);
+ }
- // Verify the ratio was NOT changed
-
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.85);
- }
+ private static void alterConfig(
+ DynamicConfigManager dynamicConfigManager, String key, String
value) throws Exception {
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(new AlterConfig(key, value,
AlterConfigOpType.SET)));
}
@Test
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index 6dd9ce7fe..b738c5a85 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -516,7 +516,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
.isInstanceOf(DiskWriteLockedException.class)
.hasMessageContaining("data disk usage");
- // recover when usage drops below (limit - 0.10) -> 0.75
+ // recover when usage drops below (limit - 0.05) -> 0.80
replicaManager.getDiskUsageMonitor().update(0.50);
assertThat(replicaManager.isDiskWriteLocked()).isFalse();
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/storage/DiskUsageMonitorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/storage/DiskUsageMonitorTest.java
index dc1291d08..8bc9b06f4 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/storage/DiskUsageMonitorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/storage/DiskUsageMonitorTest.java
@@ -26,25 +26,43 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.assertj.core.api.Assertions.within;
/** Test for {@link DiskUsageMonitor}. */
class DiskUsageMonitorTest {
private static final int SERVER_ID = 7;
+ private static final double WRITE_RECOVER_RATIO = 0.80;
@Test
- void testInvalidLimitRatioRejected() {
+ void testInvalidLimitConfigRejected() {
DiskUsageCollector collector = new
DiskUsageCollector(Collections.emptyList());
assertThatThrownBy(
() ->
new DiskUsageMonitor(
- SERVER_ID, collector, 0.0, (usage,
locked) -> {}))
+ SERVER_ID,
+ collector,
+ 0.0,
+ WRITE_RECOVER_RATIO,
+ (usage, locked) -> {}))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(
() ->
new DiskUsageMonitor(
- SERVER_ID, collector, 1.5, (usage,
locked) -> {}))
+ SERVER_ID,
+ collector,
+ 1.1,
+ WRITE_RECOVER_RATIO,
+ (usage, locked) -> {}))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(
+ () ->
+ new DiskUsageMonitor(
+ SERVER_ID, collector, 0.85, 0.0,
(usage, locked) -> {}))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(
+ () ->
+ new DiskUsageMonitor(
+ SERVER_ID, collector, 0.85, 0.85,
(usage, locked) -> {}))
.isInstanceOf(IllegalArgumentException.class);
}
@@ -81,8 +99,8 @@ class DiskUsageMonitorTest {
monitor.update(0.86);
assertThat(monitor.isLocked()).isTrue();
- // recover threshold is 0.75, 0.80 still above it -> stay locked
- monitor.update(0.80);
+ // recover threshold is 0.80, 0.81 still above it -> stay locked
+ monitor.update(0.81);
assertThat(monitor.isLocked()).isTrue();
}
@@ -94,16 +112,16 @@ class DiskUsageMonitorTest {
monitor.update(0.90);
assertThat(monitor.isLocked()).isTrue();
- monitor.update(0.75);
+ monitor.update(monitor.getWriteRecoverRatio());
assertThat(monitor.isLocked()).isFalse();
assertThat(recorder.lastLocked.get()).isFalse();
}
@Test
- void testRecoverThresholdNeverNegative() {
- DiskUsageMonitor monitor = newMonitor(0.05, new Recorder());
- assertThat(monitor.getRecoverThreshold()).isEqualTo(0.0);
- assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.05);
+ void testRecoverThresholdUsesConfiguredRecoverRatio() {
+ DiskUsageMonitor monitor = newMonitor(0.85, 0.65, new Recorder());
+ assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.85);
+ assertThat(monitor.getWriteRecoverRatio()).isEqualTo(0.65);
}
@Test
@@ -115,7 +133,8 @@ class DiskUsageMonitorTest {
Collections.singletonList(
new
File("/__fluss_disk_monitor_does_not_exist__/x")));
Recorder recorder = new Recorder();
- DiskUsageMonitor monitor = new DiskUsageMonitor(SERVER_ID, failing,
0.85, recorder);
+ DiskUsageMonitor monitor =
+ new DiskUsageMonitor(SERVER_ID, failing, 0.85,
WRITE_RECOVER_RATIO, recorder);
// First put the monitor into the locked state via update().
monitor.update(0.95);
@@ -137,7 +156,8 @@ class DiskUsageMonitorTest {
// Reuse a real collector backed by an empty data dirs list -> always
returns 0.0.
DiskUsageCollector collector = new
DiskUsageCollector(Collections.emptyList());
Recorder recorder = new Recorder();
- DiskUsageMonitor monitor = new DiskUsageMonitor(SERVER_ID, collector,
0.85, recorder);
+ DiskUsageMonitor monitor =
+ new DiskUsageMonitor(SERVER_ID, collector, 0.85,
WRITE_RECOVER_RATIO, recorder);
monitor.runOnce();
assertThat(monitor.isLocked()).isFalse();
@@ -147,66 +167,86 @@ class DiskUsageMonitorTest {
}
private DiskUsageMonitor newMonitor(double limit,
DiskUsageMonitor.Listener listener) {
+ return newMonitor(limit, WRITE_RECOVER_RATIO, listener);
+ }
+
+ private DiskUsageMonitor newMonitor(
+ double limit, double writeRecoverRatio, DiskUsageMonitor.Listener
listener) {
return new DiskUsageMonitor(
- SERVER_ID, new DiskUsageCollector(Collections.emptyList()),
limit, listener);
+ SERVER_ID,
+ new DiskUsageCollector(Collections.emptyList()),
+ limit,
+ writeRecoverRatio,
+ listener);
}
@Test
- void testUpdateWriteLimitRatioChangesThresholds() {
+ void testUpdateWriteLimitRatioKeepsRecoverRatio() {
Recorder recorder = new Recorder();
DiskUsageMonitor monitor = newMonitor(0.85, recorder);
- // Initially ratio=0.85, recover=0.75
+ // Initially ratio=0.85, recover=0.80
assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.85);
- assertThat(monitor.getRecoverThreshold()).isEqualTo(0.75);
+
assertThat(monitor.getWriteRecoverRatio()).isEqualTo(WRITE_RECOVER_RATIO);
// Simulate usage at 0.82 — should NOT lock (below 0.85)
monitor.update(0.82);
assertThat(monitor.isLocked()).isFalse();
- // Lower the limit to 0.80 — now 0.82 exceeds the new limit
- monitor.updateWriteLimitRatio(0.80);
- assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.80);
- assertThat(monitor.getRecoverThreshold()).isCloseTo(0.70,
within(1e-9));
+ // Lower the limit to 0.81; the absolute recover ratio remains 0.80.
+ monitor.updateWriteLimitRatio(0.81);
+ assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.81);
+ assertThat(monitor.getWriteRecoverRatio()).isEqualTo(0.80);
// Re-evaluate with same usage — should lock
monitor.update(0.82);
assertThat(monitor.isLocked()).isTrue();
assertThat(recorder.lastLocked.get()).isTrue();
- // Raise the limit to 0.90 and recover threshold becomes 0.80
- // 0.82 > 0.80 so should remain locked
+ // Raise the limit to 0.90; the recover ratio remains 0.80.
monitor.updateWriteLimitRatio(0.90);
- monitor.update(0.82);
+ monitor.update(0.86);
assertThat(monitor.isLocked()).isTrue();
- // Drop usage to 0.79 — now below new recover threshold 0.80, should
unlock
- monitor.update(0.79);
+ // Drop usage to 0.80; at the recover ratio, writes should resume.
+ monitor.update(0.80);
assertThat(monitor.isLocked()).isFalse();
}
+ @Test
+ void testUpdateWriteLimitConfigChangesRecoverRatio() {
+ Recorder recorder = new Recorder();
+ DiskUsageMonitor monitor = newMonitor(0.85, recorder);
+
+ monitor.updateWriteLimitConfig(0.85, 0.75);
+ assertThat(monitor.getWriteLimitRatio()).isEqualTo(0.85);
+ assertThat(monitor.getWriteRecoverRatio()).isEqualTo(0.75);
+
+ monitor.update(0.90);
+ assertThat(monitor.isLocked()).isTrue();
+ monitor.update(0.76);
+ assertThat(monitor.isLocked()).isTrue();
+ monitor.update(0.75);
+ assertThat(monitor.isLocked()).isFalse();
+ assertThat(recorder.lastLocked.get()).isFalse();
+ }
+
@Test
void testDisabledWhenRatioIsOne() {
Recorder recorder = new Recorder();
DiskUsageMonitor monitor = newMonitor(1.0, recorder);
- // Even at 100% disk usage, should NOT lock when ratio = 1.0
monitor.update(1.0);
assertThat(monitor.isLocked()).isFalse();
assertThat(recorder.lastLocked.get()).isFalse();
assertThat(recorder.lastUsage.get()).isEqualTo(1.0);
- // Any usage below 1.0 should also remain unlocked
- monitor.update(0.99);
- assertThat(monitor.isLocked()).isFalse();
-
- // If previously locked (via dynamic update), switching to 1.0 should
unlock immediately
- DiskUsageMonitor monitor2 = newMonitor(0.80, recorder);
- monitor2.update(0.85); // lock it
+ DiskUsageMonitor monitor2 = newMonitor(0.85, recorder);
+ monitor2.update(0.85);
assertThat(monitor2.isLocked()).isTrue();
monitor2.updateWriteLimitRatio(1.0);
- monitor2.update(0.85); // same usage, but ratio=1.0 -> should unlock
+ monitor2.update(0.85);
assertThat(monitor2.isLocked()).isFalse();
assertThat(recorder.lastLocked.get()).isFalse();
}
@@ -218,10 +258,16 @@ class DiskUsageMonitorTest {
assertThatThrownBy(() -> monitor.updateWriteLimitRatio(0.0))
.isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() ->
monitor.updateWriteLimitRatio(WRITE_RECOVER_RATIO))
+ .isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> monitor.updateWriteLimitRatio(1.1))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> monitor.updateWriteLimitRatio(-0.5))
.isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> monitor.updateWriteLimitConfig(0.85, 0.0))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> monitor.updateWriteLimitConfig(0.85, 0.85))
+ .isInstanceOf(IllegalArgumentException.class);
}
/** Captures the latest sample observed by the listener. */
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
index 78c5bcfd5..ec4f30426 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
@@ -358,6 +358,40 @@ class LocalDiskManagerTest {
.hasMessageContaining("None of the specified data dirs");
}
+ @Test
+ void testDiskWriteLimitConfigValidation() throws Exception {
+ File dataDir = new File(tempDir, "data-1");
+
+ try (LocalDiskManager localDiskManager =
LocalDiskManager.create(createConf(dataDir))) {
+
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.85);
+
assertThat(localDiskManager.getDiskWriteRecoverRatio()).isEqualTo(0.80);
+
assertThat(localDiskManager.getDiskUsageMonitor().getWriteRecoverRatio())
+ .isEqualTo(0.80);
+ }
+
+ Configuration disabledRatioConf = createConf(new File(tempDir,
"disabled-ratio"));
+
disabledRatioConf.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO, 1.0);
+ try (LocalDiskManager localDiskManager =
LocalDiskManager.create(disabledRatioConf)) {
+ localDiskManager.getDiskUsageMonitor().update(1.0);
+ assertThat(localDiskManager.isDiskWriteLocked()).isFalse();
+ }
+
+ Configuration invalidRatioConf = createConf(new File(tempDir,
"invalid-ratio"));
+ invalidRatioConf.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO,
1.1);
+ assertThatThrownBy(() -> LocalDiskManager.create(invalidRatioConf))
+ .isInstanceOf(IllegalConfigurationException.class);
+
+ Configuration zeroRecoverRatioConf = createConf(new File(tempDir,
"zero-recover-ratio"));
+
zeroRecoverRatioConf.set(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO,
0.0);
+ assertThatThrownBy(() -> LocalDiskManager.create(zeroRecoverRatioConf))
+ .isInstanceOf(IllegalConfigurationException.class);
+
+ Configuration equalRecoverRatioConf = createConf(new File(tempDir,
"equal-recover-ratio"));
+
equalRecoverRatioConf.set(ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO,
0.85);
+ assertThatThrownBy(() ->
LocalDiskManager.create(equalRecoverRatioConf))
+ .isInstanceOf(IllegalConfigurationException.class);
+ }
+
private Configuration createConf(File... dataDirs) {
Configuration conf = new Configuration();
conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
diff --git a/website/docs/maintenance/configuration.md
b/website/docs/maintenance/configuration.md
index fae8b45f8..a8f6d966f 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -67,22 +67,23 @@ during the Fluss cluster working.
## TabletServer
-| Option | Type | Default
| Description
[...]
-|--------------------------------------------------|------------|-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| tablet-server.id | Integer | (None)
| The id for the tablet server.
[...]
-| tablet-server.rack | String | (None)
| The rack for the TabletServer. This will be used in rack aware bucket
assignment for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`
[...]
-| data.dir | String |
/tmp/fluss-data | This configuration controls the directory where Fluss will
store its data. The default value is /tmp/fluss-data
[...]
-| server.writer-id.expiration-time | Duration | 7d
| The time that the tablet server will wait without receiving any write
request from a client before expiring the related status. The default value is
7 days.
[...]
-| server.writer-id.expiration-check-interval | Duration | 10min
| The interval at which to remove writer ids that have expired due to
`server.writer-id.expiration-time passing. The default value is 10 minutes.
[...]
-| server.background.threads | Integer | 10
| The number of threads to use for various background processing tasks. The
default value is 10.
[...]
-| server.buffer.memory-size | MemorySize | 256mb
| The total bytes of memory the server can use, e.g, buffer write-ahead-log
rows.
[...]
-| server.buffer.page-size | MemorySize | 128kb
| Size of every page in memory buffers (`server.buffer.memory-size`).
[...]
-| server.buffer.per-request-memory-size | MemorySize | 16mb
| The minimum number of bytes that will be allocated by the writer rounded
down to the closest multiple of server.buffer.page-size. It must be greater
than or equal to server.buffer.page-size. This option allows to allocate memory
in batches to have better CPU-cached friendliness due to contiguous segments.
[...]
-| server.buffer.wait-timeout | Duration | 2^(63)-1ns
| Defines how long the buffer pool will block when waiting for segments.
[...]
-| tablet-server.controlled-shutdown.max-retries | Integer | 3
| Maximum number of attempts to transfer leadership before proceeding with
an unclean shutdown during a controlled shutdown procedure.
[...]
-| tablet-server.controlled-shutdown.retry-interval | Duration | 1000ms
| Time interval between retry attempts when trying to transfer leadership
during controlled shutdown.
[...]
-| server.data-disk.write-limit-ratio | Double | 0.85
| Reject writes when the tablet server data disk usage exceeds this ratio.
Writes resume after the usage drops below `(ratio - 0.10)`. The monitor reports
the maximum usage across all distinct file stores so that a single nearly-full
disk is never masked by other low-usage disks. Set to `1.0` to disable the
disk-usage protection entirely. The valid range is `(0.1, 1.0]`. This
configuration can be updated dy [...]
-| server.data-disk.check-interval | Duration | 30s
| The interval at which the tablet server samples the local data disk usage
for the write-protection state machine. A shorter interval narrows the time
window during which writes can still flow in after the disk crosses the limit
ratio, at the cost of slightly more frequent `statvfs` calls (which are
in-memory and cheap). The default 30s is suitable for typical write workloads.
[...]
+| Option | Type | Default
| Description
[...]
+|--------------------------------------------------|------------|------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| tablet-server.id | Integer | (None)
| The id for the tablet server.
[...]
+| tablet-server.rack | String | (None)
| The rack for the TabletServer. This will be used in rack aware bucket
assignment for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`
[...]
+| data.dir | String |
/tmp/fluss-data | This configuration controls the directory where Fluss will
store its data. The default value is /tmp/fluss-data
[...]
+| server.writer-id.expiration-time | Duration | 7d
| The time that the tablet server will wait without receiving any write
request from a client before expiring the related status. The default value is
7 days.
[...]
+| server.writer-id.expiration-check-interval | Duration | 10min
| The interval at which to remove writer ids that have expired due to
`server.writer-id.expiration-time passing. The default value is 10 minutes.
[...]
+| server.background.threads | Integer | 10
| The number of threads to use for various background processing tasks. The
default value is 10.
[...]
+| server.buffer.memory-size | MemorySize | 256mb
| The total bytes of memory the server can use, e.g, buffer write-ahead-log
rows.
[...]
+| server.buffer.page-size | MemorySize | 128kb
| Size of every page in memory buffers (`server.buffer.memory-size`).
[...]
+| server.buffer.per-request-memory-size | MemorySize | 16mb
| The minimum number of bytes that will be allocated by the writer rounded
down to the closest multiple of server.buffer.page-size. It must be greater
than or equal to server.buffer.page-size. This option allows to allocate memory
in batches to have better CPU-cached friendliness due to contiguous segments.
[...]
+| server.buffer.wait-timeout | Duration | 2^(63)-1ns
| Defines how long the buffer pool will block when waiting for segments.
[...]
+| tablet-server.controlled-shutdown.max-retries | Integer | 3
| Maximum number of attempts to transfer leadership before proceeding with
an unclean shutdown during a controlled shutdown procedure.
[...]
+| tablet-server.controlled-shutdown.retry-interval | Duration | 1000ms
| Time interval between retry attempts when trying to transfer leadership
during controlled shutdown.
[...]
+| server.data-disk.write-limit-ratio | Double | 0.85
| Reject writes when the tablet server data disk usage reaches this ratio.
Writes resume when the usage reaches or drops below
`server.data-disk.write-recover-ratio`. The monitor reports the maximum usage
across all distinct file stores so that a single nearly-full disk is never
masked by other low-usage disks. Set to `1.0` to disable the disk-usage
protection entirely. The valid range is `(server.data-dis [...]
+| server.data-disk.write-recover-ratio | Double | 0.80
| Resume writes when the tablet server data disk usage reaches or drops
below this ratio. The valid range is `(0.0,
server.data-disk.write-limit-ratio)`. This configuration can be updated
dynamically without server restart.
[...]
+| server.data-disk.check-interval | Duration | 30s
| The interval at which the tablet server samples the local data disk usage
for the write-protection state machine. A shorter interval narrows the time
window during which writes can still flow in after the disk crosses the limit
ratio, at the cost of slightly more frequent `statvfs` calls (which are
in-memory and cheap). The default 30s is suitable for typical write workloads.
[...]
## Zookeeper
