You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2025/03/04 11:52:13 UTC
(flink) 02/04: [FLINK-37379][state/forst] Enable forst file cache by default
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b76108ccdb480a51eeb1c97fe6fdd57f7c06e199
Author: Zakelly <za...@gmail.com>
AuthorDate: Mon Mar 3 22:05:51 2025 +0800
[FLINK-37379][state/forst] Enable forst file cache by default
---
docs/layouts/shortcodes/generated/forst_configuration.html | 4 ++--
docs/layouts/shortcodes/generated/state_backend_forst_section.html | 4 ++--
.../src/main/java/org/apache/flink/state/forst/ForStOptions.java | 7 ++++---
.../java/org/apache/flink/state/forst/ForStResourceContainer.java | 5 +++--
.../java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java | 6 +++---
5 files changed, 14 insertions(+), 12 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/forst_configuration.html b/docs/layouts/shortcodes/generated/forst_configuration.html
index b4abb2835bd..4e859f5083d 100644
--- a/docs/layouts/shortcodes/generated/forst_configuration.html
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -28,9 +28,9 @@
</tr>
<tr>
<td><h5>state.backend.forst.cache.reserve-size</h5></td>
- <td style="word-wrap: break-word;">0 bytes</td>
+ <td style="word-wrap: break-word;">256 mb</td>
<td>MemorySize</td>
- <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit. The default value is '0 bytes', meaning the cache will not reserve any disk space.</td>
+ <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit. The default value is '256 mb', meaning the disk will be reserved that much and remaining of which can be used for cache.</td>
</tr>
<tr>
<td><h5>state.backend.forst.cache.size-based-limit</h5></td>
diff --git a/docs/layouts/shortcodes/generated/state_backend_forst_section.html b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
index f784985b60a..772de065073 100644
--- a/docs/layouts/shortcodes/generated/state_backend_forst_section.html
+++ b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
@@ -16,9 +16,9 @@
</tr>
<tr>
<td><h5>state.backend.forst.cache.reserve-size</h5></td>
- <td style="word-wrap: break-word;">0 bytes</td>
+ <td style="word-wrap: break-word;">256 mb</td>
<td>MemorySize</td>
- <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit. The default value is '0 bytes', meaning the cache will not reserve any disk space.</td>
+ <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit. The default value is '256 mb', meaning the disk will be reserved that much and remaining of which can be used for cache.</td>
</tr>
<tr>
<td><h5>state.backend.forst.cache.size-based-limit</h5></td>
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
index b61d561b532..834aab1e255 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -94,7 +94,7 @@ public class ForStOptions {
public static final ConfigOption<MemorySize> CACHE_RESERVED_SIZE =
ConfigOptions.key("state.backend.forst.cache.reserve-size")
.memoryType()
- .defaultValue(MemorySize.ZERO)
+ .defaultValue(MemorySize.ofMebiBytes(256))
.withDescription(
Description.builder()
.text(
@@ -102,9 +102,10 @@ public class ForStOptions {
+ "the cache will reserve the specified size of disk space. "
+ "This option and the '%s' option can be set simultaneously, the "
+ "smaller cache limit will be used as the upper limit. "
- + "The default value is '%s', meaning the cache will not reserve any disk space.",
+ + "The default value is '%s', meaning the disk will be reserved that much "
+ + "and remaining of which can be used for cache.",
text(CACHE_SIZE_BASE_LIMIT.key()),
- text(MemorySize.ZERO.toString()))
+ text(MemorySize.ofMebiBytes(256).toString()))
.build());
@Documentation.Section(Documentation.Sections.EXPERT_FORST)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
index 3946c38980b..5011678fcbd 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -91,9 +91,10 @@ public final class ForStResourceContainer implements AutoCloseable {
@Nullable private final Path localForStPath;
@Nullable private Path cacheBasePath;
- private long cacheCapacity;
- private long cacheReservedSize;
+ private final long cacheCapacity;
+
+ private final long cacheReservedSize;
/** The configurations from file. */
private final ReadableConfig configuration;
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
index f3f8334cc4d..7ee849d397f 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java
@@ -117,7 +117,7 @@ public class ForStFlinkFileSystem extends FileSystem implements Closeable {
long cacheReservedSize,
MetricGroup metricGroup)
throws IOException {
- if (cacheBase == null || cacheCapacity <= 0 && cacheReservedSize <= 0) {
+ if (cacheBase == null || cacheCapacity <= 0 && cacheReservedSize < 0) {
return null;
}
if (cacheBase.getFileSystem().equals(remoteForStPath.getFileSystem())) {
@@ -127,7 +127,7 @@ public class ForStFlinkFileSystem extends FileSystem implements Closeable {
return null;
}
CacheLimitPolicy cacheLimitPolicy = null;
- if (cacheCapacity > 0 && cacheReservedSize > 0) {
+ if (cacheCapacity > 0 && cacheReservedSize >= 0) {
cacheLimitPolicy =
new BundledCacheLimitPolicy(
new SizeBasedCacheLimitPolicy(cacheCapacity),
@@ -137,7 +137,7 @@ public class ForStFlinkFileSystem extends FileSystem implements Closeable {
SST_FILE_SIZE));
} else if (cacheCapacity > 0) {
cacheLimitPolicy = new SizeBasedCacheLimitPolicy(cacheCapacity);
- } else if (cacheReservedSize > 0) {
+ } else if (cacheReservedSize >= 0) {
cacheLimitPolicy =
new SpaceBasedCacheLimitPolicy(
new File(cacheBase.toString()), cacheReservedSize, SST_FILE_SIZE);