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/02/11 06:55:33 UTC
(flink) branch master updated (21e30869d65 -> dc4518fec6e)
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 21e30869d65 [FLINK-37272][python] Minor fix of FlinkAvroBufferWrapper
new 4f61d8585db [FLINK-36439][docs] Documents for State V2 APIs
new 381be9cf9ec [FLINK-36439][docs] Documents for ForSt State Backend
new 4facda6e11e [hotfix] Make ForSt state backend use checkpoint directory as its primary storage by default
new aa6a8db2def [hotfix] Rename the buffer config options for async state processing
new dc4518fec6e [FLINK-36439][docs] Documents for Disaggregated State
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content.zh/docs/deployment/config.md | 30 +-
.../datastream-v2/context_and_state_processing.md | 7 +
.../docs/dev/datastream/fault-tolerance/state.md | 2 +-
.../dev/datastream/fault-tolerance/state_v2.md} | 795 ++++++---------------
docs/content.zh/docs/ops/metrics.md | 7 +-
.../docs/ops/state/disaggregated_state.md | 193 +++++
docs/content/docs/deployment/config.md | 30 +-
.../datastream-v2/context_and_state_processing.md | 7 +
.../docs/dev/datastream/fault-tolerance/state.md | 2 +-
.../fault-tolerance/{state.md => state_v2.md} | 794 ++++++--------------
docs/content/docs/ops/metrics.md | 7 +-
docs/content/docs/ops/state/disaggregated_state.md | 193 +++++
docs/content/docs/ops/state/large_state_tuning.md | 16 +-
docs/content/docs/ops/state/state_backends.md | 57 +-
.../shortcodes/generated/expert_forst_section.html | 54 ++
.../forst_configurable_configuration.html | 168 +++++
.../shortcodes/generated/forst_configuration.html | 120 ++++
...html => forst_native_metric_configuration.html} | 122 ++--
.../generated/state_backend_forst_section.html | 66 ++
.../flink/annotation/docs/Documentation.java | 2 +
.../apache/flink/api/common/ExecutionConfig.java | 24 +-
.../flink/configuration/ExecutionOptions.java | 22 +-
flink-docs/pom.xml | 5 +
.../configuration/ConfigOptionsDocGenerator.java | 2 +
.../docs/util/ConfigurationOptionLocator.java | 3 +
.../AbstractAsyncStateStreamOperator.java | 7 +-
.../AbstractAsyncStateStreamOperatorV2.java | 6 +-
.../AbstractAsyncStateStreamOperatorTest.java | 2 +-
.../state/forst/ForStConfigurableOptions.java | 52 +-
.../state/forst/ForStNativeMetricOptions.java | 84 +--
.../org/apache/flink/state/forst/ForStOptions.java | 84 ++-
.../flink/state/forst/ForStResourceContainer.java | 6 +-
.../flink/state/forst/ForStStateBackend.java | 12 +-
.../flink/state/forst/fs/ForStFlinkFileSystem.java | 12 +-
.../state/forst/ForStStateBackendConfigTest.java | 9 +-
.../flink/state/forst/ForStStateBackendV2Test.java | 11 +-
.../flink/state/forst/ForStStateTestBase.java | 2 +-
.../StreamExecutionEnvironmentTest.java | 38 +-
38 files changed, 1676 insertions(+), 1377 deletions(-)
copy docs/{content/docs/dev/datastream/fault-tolerance/state.md => content.zh/docs/dev/datastream/fault-tolerance/state_v2.md} (51%)
create mode 100644 docs/content.zh/docs/ops/state/disaggregated_state.md
copy docs/content/docs/dev/datastream/fault-tolerance/{state.md => state_v2.md} (51%)
create mode 100644 docs/content/docs/ops/state/disaggregated_state.md
create mode 100644 docs/layouts/shortcodes/generated/expert_forst_section.html
create mode 100644 docs/layouts/shortcodes/generated/forst_configurable_configuration.html
create mode 100644 docs/layouts/shortcodes/generated/forst_configuration.html
copy docs/layouts/shortcodes/generated/{rocksdb_native_metric_configuration.html => forst_native_metric_configuration.html} (58%)
create mode 100644 docs/layouts/shortcodes/generated/state_backend_forst_section.html
(flink) 04/05: [hotfix] Rename the buffer config options for async state processing
Posted by za...@apache.org.
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 aa6a8db2def58f5ea897d0dbe34f89cc63685674
Author: Zakelly <za...@gmail.com>
AuthorDate: Sun Feb 9 23:46:41 2025 +0800
[hotfix] Rename the buffer config options for async state processing
---
.../apache/flink/api/common/ExecutionConfig.java | 24 +++++++-------
.../flink/configuration/ExecutionOptions.java | 22 ++++++-------
.../AbstractAsyncStateStreamOperator.java | 7 ++--
.../AbstractAsyncStateStreamOperatorV2.java | 6 ++--
.../AbstractAsyncStateStreamOperatorTest.java | 2 +-
.../StreamExecutionEnvironmentTest.java | 38 +++++++++++-----------
6 files changed, 50 insertions(+), 49 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index c87d333be04..4a8cf2e8c31 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -481,35 +481,35 @@ public class ExecutionConfig implements Serializable, Archiveable<ArchivedExecut
// --------------------------------------------------------------------------------------------
@Experimental
- public int getAsyncInflightRecordsLimit() {
- return configuration.get(ExecutionOptions.ASYNC_INFLIGHT_RECORDS_LIMIT);
+ public int getAsyncStateTotalBufferSize() {
+ return configuration.get(ExecutionOptions.ASYNC_STATE_TOTAL_BUFFER_SIZE);
}
@Experimental
- public ExecutionConfig setAsyncInflightRecordsLimit(int limit) {
- configuration.set(ExecutionOptions.ASYNC_INFLIGHT_RECORDS_LIMIT, limit);
+ public ExecutionConfig setAsyncStateTotalBufferSize(int limit) {
+ configuration.set(ExecutionOptions.ASYNC_STATE_TOTAL_BUFFER_SIZE, limit);
return this;
}
@Experimental
- public int getAsyncStateBufferSize() {
- return configuration.get(ExecutionOptions.ASYNC_STATE_BUFFER_SIZE);
+ public int getAsyncStateActiveBufferSize() {
+ return configuration.get(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_SIZE);
}
@Experimental
- public ExecutionConfig setAsyncStateBufferSize(int bufferSize) {
- configuration.set(ExecutionOptions.ASYNC_STATE_BUFFER_SIZE, bufferSize);
+ public ExecutionConfig setAsyncStateActiveBufferSize(int bufferSize) {
+ configuration.set(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_SIZE, bufferSize);
return this;
}
@Experimental
- public long getAsyncStateBufferTimeout() {
- return configuration.get(ExecutionOptions.ASYNC_STATE_BUFFER_TIMEOUT);
+ public long getAsyncStateActiveBufferTimeout() {
+ return configuration.get(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT);
}
@Experimental
- public ExecutionConfig setAsyncStateBufferTimeout(long timeout) {
- configuration.set(ExecutionOptions.ASYNC_STATE_BUFFER_TIMEOUT, timeout);
+ public ExecutionConfig setAsyncStateActiveBufferTimeout(long timeout) {
+ configuration.set(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT, timeout);
return this;
}
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java
index 0cec69550fb..160469a9d97 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java
@@ -194,8 +194,8 @@ public class ExecutionOptions {
@Experimental
@Documentation.ExcludeFromDocumentation(
"This is an experimental option, internal use only for now.")
- public static final ConfigOption<Integer> ASYNC_INFLIGHT_RECORDS_LIMIT =
- ConfigOptions.key("execution.async-state.in-flight-records-limit")
+ public static final ConfigOption<Integer> ASYNC_STATE_TOTAL_BUFFER_SIZE =
+ ConfigOptions.key("execution.async-state.total-buffer-size")
.intType()
.defaultValue(6000)
.withDescription(
@@ -208,36 +208,36 @@ public class ExecutionOptions {
* The size of buffer under async state execution. Async state execution provides a buffer
* mechanism to reduce state access. When the number of state requests in the buffer exceeds the
* batch size, a batched state execution would be triggered. Larger batch sizes will bring
- * higher end-to-end latency, this option works with {@link #ASYNC_STATE_BUFFER_TIMEOUT} to
- * control the frequency of triggering.
+ * higher end-to-end latency, this option works with {@link #ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT}
+ * to control the frequency of triggering.
*/
@Experimental
@Documentation.ExcludeFromDocumentation(
"This is an experimental option, internal use only for now.")
- public static final ConfigOption<Integer> ASYNC_STATE_BUFFER_SIZE =
- ConfigOptions.key("execution.async-state.buffer-size")
+ public static final ConfigOption<Integer> ASYNC_STATE_ACTIVE_BUFFER_SIZE =
+ ConfigOptions.key("execution.async-state.active-buffer-size")
.intType()
.defaultValue(1000)
.withDescription(
"The size of buffer under async state execution. Async state execution provides a buffer mechanism to reduce state access."
+ " When the number of state requests in the active buffer exceeds the batch size,"
+ " a batched state execution would be triggered. Larger batch sizes will bring higher end-to-end latency,"
- + " this option works with 'execution.async-state.buffer-timeout' to control the frequency of triggering.");
+ + " this option works with 'execution.async-state.active-buffer-timeout' to control the frequency of triggering.");
/**
* The timeout of buffer triggering in milliseconds. If the buffer has not reached the {@link
- * #ASYNC_STATE_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform
+ * #ASYNC_STATE_ACTIVE_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform
* actively.
*/
@Experimental
@Documentation.ExcludeFromDocumentation(
"This is an experimental option, internal use only for now.")
- public static final ConfigOption<Long> ASYNC_STATE_BUFFER_TIMEOUT =
- ConfigOptions.key("execution.async-state.buffer-timeout")
+ public static final ConfigOption<Long> ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT =
+ ConfigOptions.key("execution.async-state.active-buffer-timeout")
.longType()
.defaultValue(1000L)
.withDescription(
"The timeout of buffer triggering in milliseconds. If the buffer has not reached the"
- + " 'execution.async-state.buffer-size' within 'buffer-timeout' milliseconds,"
+ + " 'execution.async-state.active-buffer-size' within 'buffer-timeout' milliseconds,"
+ " a trigger will perform actively.");
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
index 94dc17a9016..df79c512a1d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
@@ -102,10 +102,11 @@ public abstract class AbstractAsyncStateStreamOperator<OUT> extends AbstractStre
final MailboxExecutor mailboxExecutor = environment.getMainMailboxExecutor();
final int maxParallelism = environment.getTaskInfo().getMaxNumberOfParallelSubtasks();
final int inFlightRecordsLimit =
- environment.getExecutionConfig().getAsyncInflightRecordsLimit();
- final int asyncBufferSize = environment.getExecutionConfig().getAsyncStateBufferSize();
+ environment.getExecutionConfig().getAsyncStateTotalBufferSize();
+ final int asyncBufferSize =
+ environment.getExecutionConfig().getAsyncStateActiveBufferSize();
final long asyncBufferTimeout =
- environment.getExecutionConfig().getAsyncStateBufferTimeout();
+ environment.getExecutionConfig().getAsyncStateActiveBufferTimeout();
this.declarationManager = new DeclarationManager();
if (isAsyncStateProcessingEnabled()) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java
index 0036055eb6a..62ab4133bcb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java
@@ -99,9 +99,9 @@ public abstract class AbstractAsyncStateStreamOperatorV2<OUT> extends AbstractSt
((DefaultKeyedStateStore) stateStore).setSupportKeyedStateApiSetV2();
}
- final int inFlightRecordsLimit = getExecutionConfig().getAsyncInflightRecordsLimit();
- final int asyncBufferSize = getExecutionConfig().getAsyncStateBufferSize();
- final long asyncBufferTimeout = getExecutionConfig().getAsyncStateBufferTimeout();
+ final int inFlightRecordsLimit = getExecutionConfig().getAsyncStateTotalBufferSize();
+ final int asyncBufferSize = getExecutionConfig().getAsyncStateActiveBufferSize();
+ final long asyncBufferTimeout = getExecutionConfig().getAsyncStateActiveBufferTimeout();
int maxParallelism = getExecutionConfig().getMaxParallelism();
this.declarationManager = new DeclarationManager();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorTest.java
index 3f29c565a4e..bb234af0eef 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorTest.java
@@ -198,7 +198,7 @@ public class AbstractAsyncStateStreamOperatorTest {
void testManyAsyncProcessWithKey() throws Exception {
// This test is for verifying AsyncExecutionController could avoid deadlock for derived
// processing requests.
- int requests = ExecutionOptions.ASYNC_INFLIGHT_RECORDS_LIMIT.defaultValue() + 1;
+ int requests = ExecutionOptions.ASYNC_STATE_TOTAL_BUFFER_SIZE.defaultValue() + 1;
TestOperatorWithMultipleDirectAsyncProcess testOperator =
new TestOperatorWithMultipleDirectAsyncProcess(ElementOrder.RECORD_ORDER, requests);
AsyncKeyedOneInputStreamOperatorTestHarness<Integer, Tuple2<Integer, String>, String>
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentTest.java
index ec89a29e1d2..295ea7d9564 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironmentTest.java
@@ -468,28 +468,28 @@ class StreamExecutionEnvironmentTest {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.configure(config, this.getClass().getClassLoader());
- assertThat(env.getConfig().getAsyncInflightRecordsLimit())
- .isEqualTo(ExecutionOptions.ASYNC_INFLIGHT_RECORDS_LIMIT.defaultValue());
- assertThat(env.getConfig().getAsyncStateBufferSize())
- .isEqualTo(ExecutionOptions.ASYNC_STATE_BUFFER_SIZE.defaultValue());
- assertThat(env.getConfig().getAsyncStateBufferTimeout())
- .isEqualTo(ExecutionOptions.ASYNC_STATE_BUFFER_TIMEOUT.defaultValue());
-
- config.set(ExecutionOptions.ASYNC_INFLIGHT_RECORDS_LIMIT, 3);
- config.set(ExecutionOptions.ASYNC_STATE_BUFFER_SIZE, 2);
- config.set(ExecutionOptions.ASYNC_STATE_BUFFER_TIMEOUT, 1L);
+ assertThat(env.getConfig().getAsyncStateTotalBufferSize())
+ .isEqualTo(ExecutionOptions.ASYNC_STATE_TOTAL_BUFFER_SIZE.defaultValue());
+ assertThat(env.getConfig().getAsyncStateActiveBufferSize())
+ .isEqualTo(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_SIZE.defaultValue());
+ assertThat(env.getConfig().getAsyncStateActiveBufferTimeout())
+ .isEqualTo(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT.defaultValue());
+
+ config.set(ExecutionOptions.ASYNC_STATE_TOTAL_BUFFER_SIZE, 3);
+ config.set(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_SIZE, 2);
+ config.set(ExecutionOptions.ASYNC_STATE_ACTIVE_BUFFER_TIMEOUT, 1L);
env.configure(config, this.getClass().getClassLoader());
- assertThat(env.getConfig().getAsyncInflightRecordsLimit()).isEqualTo(3);
- assertThat(env.getConfig().getAsyncStateBufferSize()).isEqualTo(2);
- assertThat(env.getConfig().getAsyncStateBufferTimeout()).isEqualTo(1);
+ assertThat(env.getConfig().getAsyncStateTotalBufferSize()).isEqualTo(3);
+ assertThat(env.getConfig().getAsyncStateActiveBufferSize()).isEqualTo(2);
+ assertThat(env.getConfig().getAsyncStateActiveBufferTimeout()).isEqualTo(1);
env.getConfig()
- .setAsyncInflightRecordsLimit(6)
- .setAsyncStateBufferSize(5)
- .setAsyncStateBufferTimeout(4);
- assertThat(env.getConfig().getAsyncInflightRecordsLimit()).isEqualTo(6);
- assertThat(env.getConfig().getAsyncStateBufferSize()).isEqualTo(5);
- assertThat(env.getConfig().getAsyncStateBufferTimeout()).isEqualTo(4);
+ .setAsyncStateTotalBufferSize(6)
+ .setAsyncStateActiveBufferSize(5)
+ .setAsyncStateActiveBufferTimeout(4);
+ assertThat(env.getConfig().getAsyncStateTotalBufferSize()).isEqualTo(6);
+ assertThat(env.getConfig().getAsyncStateActiveBufferSize()).isEqualTo(5);
+ assertThat(env.getConfig().getAsyncStateActiveBufferTimeout()).isEqualTo(4);
}
private void testBufferTimeout(Configuration config, StreamExecutionEnvironment env) {
(flink) 05/05: [FLINK-36439][docs] Documents for Disaggregated State
Posted by za...@apache.org.
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 dc4518fec6eb86fff38db7a59359a0239c95cc77
Author: Zakelly <za...@gmail.com>
AuthorDate: Sun Feb 9 23:58:52 2025 +0800
[FLINK-36439][docs] Documents for Disaggregated State
---
.../dev/datastream/fault-tolerance/state_v2.md | 4 +-
.../docs/ops/state/disaggregated_state.md | 193 +++++++++++++++++++++
.../dev/datastream/fault-tolerance/state_v2.md | 4 +-
docs/content/docs/ops/state/disaggregated_state.md | 193 +++++++++++++++++++++
docs/content/docs/ops/state/state_backends.md | 2 +-
5 files changed, 393 insertions(+), 3 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
index 6ef16802f80..ee46a2ab8ae 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
@@ -35,7 +35,9 @@ to learn about the concepts behind stateful stream processing.
The new state API is designed to be more flexible than the previous API. User can perform
asynchronous state operations, thus making it more powerful and more efficient.
The asynchronous state access is essential for the state backend to be able to handle
-large state sizes and to be able to spill to remote file systems when necessary.
+large state sizes and to be able to spill to remote file systems when necessary.
+This is called the 'disaggregated state management'. For more information about this,
+please see [Disaggregated State Management]({{< ref "docs/ops/state/disaggregated_state" >}}).
## Keyed DataStream
diff --git a/docs/content.zh/docs/ops/state/disaggregated_state.md b/docs/content.zh/docs/ops/state/disaggregated_state.md
new file mode 100644
index 00000000000..146de00094a
--- /dev/null
+++ b/docs/content.zh/docs/ops/state/disaggregated_state.md
@@ -0,0 +1,193 @@
+---
+title: "Disaggregated State Management"
+weight: 20
+type: docs
+aliases:
+ - /ops/state/disaggregated_state.html
+ - /apis/streaming/disaggregated_state.html
+---
+<!--
+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.
+-->
+
+# Disaggregated State Management
+
+## Overview
+
+For the first ten years of Flink, the state management is based on memory or local disk of the TaskManager.
+This approach works well for most use cases, but it has some limitations:
+ * **Local Disk Constraints**: The state size is limited by the memory or disk size of the TaskManager.
+ * **Spiky Resource Usage**: The local state model triggers periodic CPU and network I/O bursts during checkpointing or SST files compaction.
+ * **Heavy Recovery**: State needs to be downloaded during recovery. The recovery time is
+proportional to the state size, which can be slow for large state sizes.
+
+In Flink 2.0, we introduced the disaggregated state management. This feature allows users to store
+the state in external storage systems like S3, HDFS, etc. This is useful when the state size
+is extremely large. It could be used to store the state in a more cost-effective way, or to
+persist or recovery the state in a more lightweight way. The benefits of disaggregated state management are:
+ * **Unlimited State Size**: The state size is only limited by the external storage system.
+ * **Stable Resource Usage**: The state is stored in external storage, thus the checkpoint could be very lightweight.
+And the SST files compaction could be done remotely (TODO).
+ * **Fast Recovery**: No need to download the state during recovery. The recovery time is
+independent of the state size.
+ * **Flexible**: Users can easily choose different external storage systems or I/O performance levels,
+or scale the storage based on their requirements without change their hardware.
+ * **Cost-effective**: External storage are usually cheaper than local disk. Users can flexibly
+adjust computing resources and storage resources independently if there is any bottleneck.
+
+The disaggregated state management contains three parts:
+ * **ForSt State Backend**: A state backend that stores the state in external storage systems. It
+can also leverage the local disk for caching and buffering. The asynchronous I/O model is used to
+read and write the state. For more details, see [ForSt State Backend]({{< ref "docs/ops/state/state_backends#the-forststatebackend" >}}).
+ * **New State APIs**: The new state APIs (State V2) are introduced to perform asynchronous state
+reads and writes, which is essential for overcoming the high network latency when accessing
+the disaggregated state. For more details, see [New State APIs]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+ * **SQL Support**: Many SQL operators are rewritten to support the disaggregated state management
+and asynchronous state access. User can easily enable these by setting the configuration.
+
+{{ <hint info >}}
+Disaggregated state and asynchronous state access are encouraged for large state. However, when
+the state size is small, the local state management with synchronous state access is a better
+choice.
+{{ </hint >}}
+
+{{ <hint info >}}
+The disaggregated state management is still in experimental state. We are working on improving
+the performance and stability of this feature. The APIs and configurations may change in future
+release.
+{{ </hint >}}
+
+## Quick Start
+
+### For SQL Jobs
+
+To enable the disaggregated state management in SQL jobs, you can set the following configurations:
+```yaml
+state.backend.type: forst
+table.exec.async-state.enabled: true
+
+# enable checkpoints, checkpoint directory is required
+execution.checkpointing.incremental: true
+execution.checkpointing.dir: s3://your-bucket/flink-checkpoints
+
+# We don't support the mini-batch and two-phase aggregation in asynchronous state access yet.
+table.exec.mini-batch.enabled: false
+table.optimizer.agg-phase-strategy: ONE_PHASE
+```
+Thus, you could leverage the disaggregated state management and asynchronous state access in
+your SQL jobs. We haven't implemented the full support for the asynchronous state access
+in SQL yet. If the SQL operators you are using are not supported, the operator will fall back
+to the synchronous state implementation automatically. The performance may not be optimal in
+this case. The supported stateful operators are:
+ - Rank (Top1, Append TopN)
+ - Row Time Deduplicate
+ - Aggregate (without distinct)
+ - Join
+ - Window Join
+ - Tumble / Hop / Cumulative Window Aggregate
+
+
+### For DataStream Jobs
+
+To enable the disaggregated state management in DataStream jobs, firstly you should use
+the `ForStStateBackend`. Configure via code in per-job mode:
+```java
+Configuration config = new Configuration();
+config.set(StateBackendOptions.STATE_BACKEND, "forst");
+config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "s3://your-bucket/flink-checkpoints");
+config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
+env.configure(config);
+```
+Or configure via `config.yaml`:
+```yaml
+state.backend.type: forst
+
+# enable checkpoints, checkpoint directory is required
+execution.checkpointing.incremental: true
+execution.checkpointing.dir: s3://your-bucket/flink-checkpoints
+```
+
+Then, you should write your datastream jobs with the new state APIs. For more
+details, see [State V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+
+## Advanced Tuning Options
+
+### Tuning ForSt State Backend
+
+The `ForStStateBackend` has many configurations to tune the performance.
+The design of ForSt is very similar to RocksDB, and the configurable options are almost the same,
+so you can refer to [large state tuning]({{< ref "docs/ops/state/large_state_tuning#tuning-rocksdb-or-forst" >}})
+to tune the ForSt state backend.
+
+Besides that, the following sections introduce some unique configurations for ForSt.
+
+#### ForSt Primary Storage Location
+
+By default, ForSt stores the state in the checkpoint directory. In this case,
+ForSt could perform lightweight checkpoints and fast recovery. However, users may
+want to store the state in a different location, e.g., a different bucket in S3.
+You can set the following configuration to specify the primary storage location:
+```yaml
+state.backend.forst.primary-dir: s3://your-bucket/forst-state
+```
+
+**Note**: If you set this configuration, you may not be able to leverage the lightweight
+checkpoint and fast recovery, since the ForSt will perform file copy between the primary
+storage location and the checkpoint directory during checkpointing and recovery.
+
+
+#### ForSt File Cache
+
+ForSt uses the local disk for caching and buffering. The granularity of the cache is whole file.
+This is enabled by default, except when the primary storage location is set to local.
+There are two capacity limit policies for the cache:
+ - Size-based: The cache will evict the oldest files when the cache size exceeds the limit.
+ - Reserved-based: The cache will evict the oldest files when the reserved space on disk
+(the disk where cache directory is) is not enough.
+Corresponding configurations are:
+```yaml
+state.backend.forst.cache.size-based-limit: 1GB
+state.backend.forst.cache.reserve-size: 10GB
+```
+Those can take effect together. If so, the cache will evict the oldest files when the cache
+size exceeds either the size-based limit or the reserved size limit.
+
+One can also specify the cache directory via:
+```yaml
+state.backend.forst.cache.dir: /tmp/forst-cache
+```
+
+#### ForSt Asynchronous Threads
+
+ForSt uses asynchronous I/O to read and write the state. There are three types of threads:
+ - Coordinator thread: The thread that coordinates the asynchronous read and write.
+ - Read thread: The thread that reads the state asynchronously.
+ - Write thread: The thread that writes the state asynchronously.
+
+The number of asynchronous threads is configurable. Typically, you don't need to adjust these
+values since the default values are good enough for most cases.
+In case for special needs, you can set the following configuration to specify the number of
+asynchronous threads:
+ - `state.backend.forst.executor.read-io-parallelism`: The number of asynchronous threads for read. Default is 3.
+ - `state.backend.forst.executor.write-io-parallelism`: The number of asynchronous threads for write. Default is 1.
+ - `state.backend.forst.executor.inline-write`: Whether to inline the write operation in the coordinator thread.
+Default is true. Setting this to false will raise the CPU usage.
+ - `state.backend.forst.executor.inline-coordinator`: Whether to let task thread be the coordinator thread.
+Default is true. Setting this to false will raise the CPU usage.
+
+{{ <top> }}
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md b/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
index d1abd47ae49..a63ae36c001 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
@@ -35,7 +35,9 @@ to learn about the concepts behind stateful stream processing.
The new state API is designed to be more flexible than the previous API. User can perform
asynchronous state operations, thus making it more powerful and more efficient.
The asynchronous state access is essential for the state backend to be able to handle
-large state sizes and to be able to spill to remote file systems when necessary.
+large state sizes and to be able to spill to remote file systems when necessary.
+This is called the 'disaggregated state management'. For more information about this,
+please see [Disaggregated State Management]({{< ref "docs/ops/state/disaggregated_state" >}}).
## Keyed DataStream
diff --git a/docs/content/docs/ops/state/disaggregated_state.md b/docs/content/docs/ops/state/disaggregated_state.md
new file mode 100644
index 00000000000..146de00094a
--- /dev/null
+++ b/docs/content/docs/ops/state/disaggregated_state.md
@@ -0,0 +1,193 @@
+---
+title: "Disaggregated State Management"
+weight: 20
+type: docs
+aliases:
+ - /ops/state/disaggregated_state.html
+ - /apis/streaming/disaggregated_state.html
+---
+<!--
+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.
+-->
+
+# Disaggregated State Management
+
+## Overview
+
+For the first ten years of Flink, the state management is based on memory or local disk of the TaskManager.
+This approach works well for most use cases, but it has some limitations:
+ * **Local Disk Constraints**: The state size is limited by the memory or disk size of the TaskManager.
+ * **Spiky Resource Usage**: The local state model triggers periodic CPU and network I/O bursts during checkpointing or SST files compaction.
+ * **Heavy Recovery**: State needs to be downloaded during recovery. The recovery time is
+proportional to the state size, which can be slow for large state sizes.
+
+In Flink 2.0, we introduced the disaggregated state management. This feature allows users to store
+the state in external storage systems like S3, HDFS, etc. This is useful when the state size
+is extremely large. It could be used to store the state in a more cost-effective way, or to
+persist or recovery the state in a more lightweight way. The benefits of disaggregated state management are:
+ * **Unlimited State Size**: The state size is only limited by the external storage system.
+ * **Stable Resource Usage**: The state is stored in external storage, thus the checkpoint could be very lightweight.
+And the SST files compaction could be done remotely (TODO).
+ * **Fast Recovery**: No need to download the state during recovery. The recovery time is
+independent of the state size.
+ * **Flexible**: Users can easily choose different external storage systems or I/O performance levels,
+or scale the storage based on their requirements without change their hardware.
+ * **Cost-effective**: External storage are usually cheaper than local disk. Users can flexibly
+adjust computing resources and storage resources independently if there is any bottleneck.
+
+The disaggregated state management contains three parts:
+ * **ForSt State Backend**: A state backend that stores the state in external storage systems. It
+can also leverage the local disk for caching and buffering. The asynchronous I/O model is used to
+read and write the state. For more details, see [ForSt State Backend]({{< ref "docs/ops/state/state_backends#the-forststatebackend" >}}).
+ * **New State APIs**: The new state APIs (State V2) are introduced to perform asynchronous state
+reads and writes, which is essential for overcoming the high network latency when accessing
+the disaggregated state. For more details, see [New State APIs]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+ * **SQL Support**: Many SQL operators are rewritten to support the disaggregated state management
+and asynchronous state access. User can easily enable these by setting the configuration.
+
+{{ <hint info >}}
+Disaggregated state and asynchronous state access are encouraged for large state. However, when
+the state size is small, the local state management with synchronous state access is a better
+choice.
+{{ </hint >}}
+
+{{ <hint info >}}
+The disaggregated state management is still in experimental state. We are working on improving
+the performance and stability of this feature. The APIs and configurations may change in future
+release.
+{{ </hint >}}
+
+## Quick Start
+
+### For SQL Jobs
+
+To enable the disaggregated state management in SQL jobs, you can set the following configurations:
+```yaml
+state.backend.type: forst
+table.exec.async-state.enabled: true
+
+# enable checkpoints, checkpoint directory is required
+execution.checkpointing.incremental: true
+execution.checkpointing.dir: s3://your-bucket/flink-checkpoints
+
+# We don't support the mini-batch and two-phase aggregation in asynchronous state access yet.
+table.exec.mini-batch.enabled: false
+table.optimizer.agg-phase-strategy: ONE_PHASE
+```
+Thus, you could leverage the disaggregated state management and asynchronous state access in
+your SQL jobs. We haven't implemented the full support for the asynchronous state access
+in SQL yet. If the SQL operators you are using are not supported, the operator will fall back
+to the synchronous state implementation automatically. The performance may not be optimal in
+this case. The supported stateful operators are:
+ - Rank (Top1, Append TopN)
+ - Row Time Deduplicate
+ - Aggregate (without distinct)
+ - Join
+ - Window Join
+ - Tumble / Hop / Cumulative Window Aggregate
+
+
+### For DataStream Jobs
+
+To enable the disaggregated state management in DataStream jobs, firstly you should use
+the `ForStStateBackend`. Configure via code in per-job mode:
+```java
+Configuration config = new Configuration();
+config.set(StateBackendOptions.STATE_BACKEND, "forst");
+config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "s3://your-bucket/flink-checkpoints");
+config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
+env.configure(config);
+```
+Or configure via `config.yaml`:
+```yaml
+state.backend.type: forst
+
+# enable checkpoints, checkpoint directory is required
+execution.checkpointing.incremental: true
+execution.checkpointing.dir: s3://your-bucket/flink-checkpoints
+```
+
+Then, you should write your datastream jobs with the new state APIs. For more
+details, see [State V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+
+## Advanced Tuning Options
+
+### Tuning ForSt State Backend
+
+The `ForStStateBackend` has many configurations to tune the performance.
+The design of ForSt is very similar to RocksDB, and the configurable options are almost the same,
+so you can refer to [large state tuning]({{< ref "docs/ops/state/large_state_tuning#tuning-rocksdb-or-forst" >}})
+to tune the ForSt state backend.
+
+Besides that, the following sections introduce some unique configurations for ForSt.
+
+#### ForSt Primary Storage Location
+
+By default, ForSt stores the state in the checkpoint directory. In this case,
+ForSt could perform lightweight checkpoints and fast recovery. However, users may
+want to store the state in a different location, e.g., a different bucket in S3.
+You can set the following configuration to specify the primary storage location:
+```yaml
+state.backend.forst.primary-dir: s3://your-bucket/forst-state
+```
+
+**Note**: If you set this configuration, you may not be able to leverage the lightweight
+checkpoint and fast recovery, since the ForSt will perform file copy between the primary
+storage location and the checkpoint directory during checkpointing and recovery.
+
+
+#### ForSt File Cache
+
+ForSt uses the local disk for caching and buffering. The granularity of the cache is whole file.
+This is enabled by default, except when the primary storage location is set to local.
+There are two capacity limit policies for the cache:
+ - Size-based: The cache will evict the oldest files when the cache size exceeds the limit.
+ - Reserved-based: The cache will evict the oldest files when the reserved space on disk
+(the disk where cache directory is) is not enough.
+Corresponding configurations are:
+```yaml
+state.backend.forst.cache.size-based-limit: 1GB
+state.backend.forst.cache.reserve-size: 10GB
+```
+Those can take effect together. If so, the cache will evict the oldest files when the cache
+size exceeds either the size-based limit or the reserved size limit.
+
+One can also specify the cache directory via:
+```yaml
+state.backend.forst.cache.dir: /tmp/forst-cache
+```
+
+#### ForSt Asynchronous Threads
+
+ForSt uses asynchronous I/O to read and write the state. There are three types of threads:
+ - Coordinator thread: The thread that coordinates the asynchronous read and write.
+ - Read thread: The thread that reads the state asynchronously.
+ - Write thread: The thread that writes the state asynchronously.
+
+The number of asynchronous threads is configurable. Typically, you don't need to adjust these
+values since the default values are good enough for most cases.
+In case for special needs, you can set the following configuration to specify the number of
+asynchronous threads:
+ - `state.backend.forst.executor.read-io-parallelism`: The number of asynchronous threads for read. Default is 3.
+ - `state.backend.forst.executor.write-io-parallelism`: The number of asynchronous threads for write. Default is 1.
+ - `state.backend.forst.executor.inline-write`: Whether to inline the write operation in the coordinator thread.
+Default is true. Setting this to false will raise the CPU usage.
+ - `state.backend.forst.executor.inline-coordinator`: Whether to let task thread be the coordinator thread.
+Default is true. Setting this to false will raise the CPU usage.
+
+{{ <top> }}
diff --git a/docs/content/docs/ops/state/state_backends.md b/docs/content/docs/ops/state/state_backends.md
index b3cb7462333..05e35e87bd0 100644
--- a/docs/content/docs/ops/state/state_backends.md
+++ b/docs/content/docs/ops/state/state_backends.md
@@ -99,7 +99,7 @@ The total memory amount of RocksDB instance(s) per slot can also be bounded, ple
The *ForStStateBackend* is a state backend that is based on [ForSt project](https://github.com/ververica/ForSt),
which is also a LSM-tree structured key-value store and built on top of the RocksDB.
-It is designed to provide a more efficient way to store and access state in Flink applications.
+It is designed for disaggregated state management, for more details, see [here]({{< ref "docs/ops/state/disaggregated_state" >}}).
Most importantly, it can hold its sst files on remote file systems that Flink supports, such as HDFS, S3, etc.
This allows Flink to scale the state size beyond the local disk capacity of the TaskManager.
Moreover, by putting the sst files on remote file systems, it can also provide a more lightweight
(flink) 03/05: [hotfix] Make ForSt state backend use checkpoint directory as its primary storage by default
Posted by za...@apache.org.
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 4facda6e11edca7e269b95456ee75b25d2f10492
Author: Zakelly <za...@gmail.com>
AuthorDate: Sat Feb 8 19:56:21 2025 +0800
[hotfix] Make ForSt state backend use checkpoint directory as its primary storage by default
---
.../shortcodes/generated/forst_configuration.html | 8 ++++----
.../generated/state_backend_forst_section.html | 8 ++++----
.../org/apache/flink/state/forst/ForStOptions.java | 20 ++++++++++++--------
.../flink/state/forst/ForStResourceContainer.java | 6 +++++-
.../apache/flink/state/forst/ForStStateBackend.java | 12 ++++++++----
.../flink/state/forst/fs/ForStFlinkFileSystem.java | 12 +++++++++++-
.../state/forst/ForStStateBackendConfigTest.java | 9 ++++++---
.../flink/state/forst/ForStStateBackendV2Test.java | 11 +++++++----
.../apache/flink/state/forst/ForStStateTestBase.java | 2 +-
9 files changed, 58 insertions(+), 30 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/forst_configuration.html b/docs/layouts/shortcodes/generated/forst_configuration.html
index 56d6855ad82..c0ab2ef24a2 100644
--- a/docs/layouts/shortcodes/generated/forst_configuration.html
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -12,7 +12,7 @@
<td><h5>state.backend.forst.cache.dir</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ <td>The directory where ForSt caches its SST files, fallback to the subdirectory of '/cache' under the value of 'state.backend.forst.local-dir' if not configured.</td>
</tr>
<tr>
<td><h5>state.backend.forst.cache.reserve-size</h5></td>
@@ -99,10 +99,10 @@
<td>The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions after loading ones from 'ForStConfigurableOptions' and pre-defined options.</td>
</tr>
<tr>
- <td><h5>state.backend.forst.remote-dir</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td><h5>state.backend.forst.primary-dir</h5></td>
+ <td style="word-wrap: break-word;">"checkpoint-dir"</td>
<td>String</td>
- <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ <td>The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. Recognized shortcut name is 'checkpoint-dir', which means that ForSt shares the directory with checkpoint, and 'local-dir', which means that ForSt will use the local directory of TaskManager.</td>
</tr>
<tr>
<td><h5>state.backend.forst.timer-service.cache-size</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 29459c4157d..38505dbd215 100644
--- a/docs/layouts/shortcodes/generated/state_backend_forst_section.html
+++ b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
@@ -12,7 +12,7 @@
<td><h5>state.backend.forst.cache.dir</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ <td>The directory where ForSt caches its SST files, fallback to the subdirectory of '/cache' under the value of 'state.backend.forst.local-dir' if not configured.</td>
</tr>
<tr>
<td><h5>state.backend.forst.cache.reserve-size</h5></td>
@@ -45,10 +45,10 @@
<td>With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. The partitioned index/filter then uses the top-level index to load on demand into the block cache the partitions that are required to perform the index/filter query. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.mem [...]
</tr>
<tr>
- <td><h5>state.backend.forst.remote-dir</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td><h5>state.backend.forst.primary-dir</h5></td>
+ <td style="word-wrap: break-word;">"checkpoint-dir"</td>
<td>String</td>
- <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ <td>The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. Recognized shortcut name is 'checkpoint-dir', which means that ForSt shares the directory with checkpoint, and 'local-dir', which means that ForSt will use the local directory of TaskManager.</td>
</tr>
<tr>
<td><h5>state.backend.forst.timer-service.cache-size</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 8c38ed83038..5eb8c1cebda 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
@@ -27,8 +27,9 @@ import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.description.Description;
import org.apache.flink.configuration.description.TextElement;
+import static org.apache.flink.state.forst.ForStStateBackend.CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT;
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
import static org.apache.flink.state.forst.ForStStateBackend.PriorityQueueStateType.ForStDB;
-import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHECKPOINT;
/** Configuration options for the ForStStateBackend. */
@Experimental
@@ -52,15 +53,17 @@ public class ForStOptions {
/** The remote directory where ForSt puts its SST files. */
@Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
- public static final ConfigOption<String> REMOTE_DIRECTORY =
- ConfigOptions.key("state.backend.forst.remote-dir")
+ public static final ConfigOption<String> PRIMARY_DIRECTORY =
+ ConfigOptions.key("state.backend.forst.primary-dir")
.stringType()
- .noDefaultValue()
+ .defaultValue(CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT)
.withDescription(
String.format(
- "The remote directory where ForSt puts its SST files, fallback to %s if not configured."
- + " Recognized shortcut name is '%s', which means that forst shares the directory with checkpoint.",
- LOCAL_DIRECTORIES.key(), REMOTE_SHORTCUT_CHECKPOINT));
+ "The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. "
+ + "Recognized shortcut name is '%s', which means that ForSt shares the directory with checkpoint, "
+ + "and '%s', which means that ForSt will use the local directory of TaskManager.",
+ CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT,
+ LOCAL_DIR_AS_PRIMARY_SHORTCUT));
@Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<String> CACHE_DIRECTORY =
@@ -69,7 +72,8 @@ public class ForStOptions {
.noDefaultValue()
.withDescription(
String.format(
- "The directory where ForSt caches its SST files, fallback to %s/cache if not configured.",
+ "The directory where ForSt caches its SST files, fallback to the "
+ + "subdirectory of '/cache' under the value of '%s' if not configured.",
LOCAL_DIRECTORIES.key()));
@Documentation.Section(Documentation.Sections.STATE_BACKEND_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 4f8a36dabb7..4b6bf545bd1 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
@@ -386,7 +386,11 @@ public final class ForStResourceContainer implements AutoCloseable {
remoteForStPath.toUri(),
localForStPath,
ForStFlinkFileSystem.getFileBasedCache(
- cacheBasePath, cacheCapacity, cacheReservedSize, metricGroup));
+ cacheBasePath,
+ remoteForStPath,
+ cacheCapacity,
+ cacheReservedSize,
+ metricGroup));
} else {
forStFileSystem = null;
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
index 64f6c57c1fa..105b213b94e 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java
@@ -95,7 +95,8 @@ import static org.apache.flink.util.Preconditions.checkArgument;
public class ForStStateBackend extends AbstractManagedMemoryStateBackend
implements ConfigurableStateBackend {
- public static final String REMOTE_SHORTCUT_CHECKPOINT = "checkpoint-dir";
+ public static final String CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT = "checkpoint-dir";
+ public static final String LOCAL_DIR_AS_PRIMARY_SHORTCUT = "local-dir";
private static final long serialVersionUID = 1L;
@@ -216,12 +217,15 @@ public class ForStStateBackend extends AbstractManagedMemoryStateBackend
if (original.remoteForStDirectory != null) {
this.remoteForStDirectory = original.remoteForStDirectory;
} else {
- String remoteDirStr = config.get(ForStOptions.REMOTE_DIRECTORY);
- if (REMOTE_SHORTCUT_CHECKPOINT.equals(remoteDirStr)) {
+ String remoteDirStr = config.get(ForStOptions.PRIMARY_DIRECTORY);
+ if (CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT.equals(remoteDirStr)) {
this.remoteForStDirectory = null;
this.remoteShareWithCheckpoint = true;
} else {
- this.remoteForStDirectory = remoteDirStr == null ? null : new Path(remoteDirStr);
+ this.remoteForStDirectory =
+ remoteDirStr == null || LOCAL_DIR_AS_PRIMARY_SHORTCUT.equals(remoteDirStr)
+ ? null
+ : new Path(remoteDirStr);
}
}
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 3966926397c..aea5c3d4d43 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
@@ -108,11 +108,21 @@ public class ForStFlinkFileSystem extends FileSystem {
}
public static FileBasedCache getFileBasedCache(
- Path cacheBase, long cacheCapacity, long cacheReservedSize, MetricGroup metricGroup)
+ Path cacheBase,
+ Path remoteForStPath,
+ long cacheCapacity,
+ long cacheReservedSize,
+ MetricGroup metricGroup)
throws IOException {
if (cacheBase == null || cacheCapacity <= 0 && cacheReservedSize <= 0) {
return null;
}
+ if (cacheBase.getFileSystem().equals(remoteForStPath.getFileSystem())) {
+ LOG.info(
+ "Skip creating ForSt cache "
+ + "since the cache and primary path are on the same file system.");
+ return null;
+ }
CacheLimitPolicy cacheLimitPolicy = null;
if (cacheCapacity > 0 && cacheReservedSize > 0) {
cacheLimitPolicy =
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
index ce0897f864b..91183e551c9 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java
@@ -69,6 +69,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
import static org.apache.flink.state.forst.ForStTestUtils.createKeyedStateBackend;
import static org.hamcrest.CoreMatchers.anyOf;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -283,6 +284,7 @@ public class ForStStateBackendConfigTest {
conf.set(ForStConfigurableOptions.LOG_LEVEL, InfoLogLevel.DEBUG_LEVEL);
conf.set(ForStConfigurableOptions.LOG_FILE_NUM, 4);
conf.set(ForStConfigurableOptions.LOG_MAX_FILE_SIZE, MemorySize.parse("1kb"));
+ conf.set(ForStOptions.PRIMARY_DIRECTORY, LOCAL_DIR_AS_PRIMARY_SHORTCUT);
final ForStStateBackend forStBackend =
new ForStStateBackend().configure(conf, getClass().getClassLoader());
final String dbStoragePath = new Path(folder.toURI().toString()).toString();
@@ -742,10 +744,11 @@ public class ForStStateBackendConfigTest {
}
@Test
- public void testRemoteDirectory() throws Exception {
+ public void testPrimaryDirectory() throws Exception {
FileSystem.initialize(new Configuration(), null);
Configuration configuration = new Configuration();
- configuration.set(ForStOptions.REMOTE_DIRECTORY, tempFolder.newFolder().toURI().toString());
+ configuration.set(
+ ForStOptions.PRIMARY_DIRECTORY, tempFolder.newFolder().toURI().toString());
ForStStateBackend forStStateBackend =
new ForStStateBackend().configure(configuration, null);
ForStKeyedStateBackend<Integer> keyedBackend = null;
@@ -759,7 +762,7 @@ public class ForStStateBackendConfigTest {
keyedBackend
.getRemoteBasePath()
.toString()
- .startsWith(configuration.get(ForStOptions.REMOTE_DIRECTORY)));
+ .startsWith(configuration.get(ForStOptions.PRIMARY_DIRECTORY)));
} finally {
if (keyedBackend != null) {
keyedBackend.dispose();
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
index 0ac3f35b748..60dff57672d 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java
@@ -54,8 +54,9 @@ import static org.apache.flink.configuration.StateRecoveryOptions.RESTORE_MODE;
import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING;
import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
import static org.apache.flink.state.forst.ForStOptions.LOCAL_DIRECTORIES;
-import static org.apache.flink.state.forst.ForStOptions.REMOTE_DIRECTORY;
-import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHECKPOINT;
+import static org.apache.flink.state.forst.ForStOptions.PRIMARY_DIRECTORY;
+import static org.apache.flink.state.forst.ForStStateBackend.CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT;
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
/** Tests for the async keyed state backend part of {@link ForStStateBackend}. */
@@ -169,7 +170,9 @@ class ForStStateBackendV2Test extends StateBackendTestV2Base<ForStStateBackend>
config.set(LOCAL_DIRECTORIES, tempFolderForForStLocal.toString());
}
if (hasRemoteDir) {
- config.set(REMOTE_DIRECTORY, tempFolderForForstRemote.toString());
+ config.set(PRIMARY_DIRECTORY, tempFolderForForstRemote.toString());
+ } else {
+ config.set(PRIMARY_DIRECTORY, LOCAL_DIR_AS_PRIMARY_SHORTCUT);
}
config.set(USE_INGEST_DB_RESTORE_MODE, useIngestDbRestoreMode);
config.set(USE_DELETE_FILES_IN_RANGE_DURING_RESCALING, useDeleteFileInRange);
@@ -207,7 +210,7 @@ class ForStStateBackendV2Test extends StateBackendTestV2Base<ForStStateBackend>
Configuration config = new Configuration();
config.set(LOCAL_DIRECTORIES, tempFolderForForStLocal.toString());
- config.set(REMOTE_DIRECTORY, REMOTE_SHORTCUT_CHECKPOINT);
+ config.set(PRIMARY_DIRECTORY, CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT);
config.set(CheckpointingOptions.CREATE_CHECKPOINT_SUB_DIR, createJob);
checkpointStorage =
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
index cd6cf7a95d0..f888546bd94 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java
@@ -61,7 +61,7 @@ public class ForStStateTestBase {
public void setup(@TempDir File temporaryFolder) throws IOException {
FileSystem.initialize(new Configuration(), null);
Configuration configuration = new Configuration();
- configuration.set(ForStOptions.REMOTE_DIRECTORY, temporaryFolder.toURI().toString());
+ configuration.set(ForStOptions.PRIMARY_DIRECTORY, temporaryFolder.toURI().toString());
ForStStateBackend forStStateBackend =
new ForStStateBackend().configure(configuration, null);
(flink) 01/05: [FLINK-36439][docs] Documents for State V2 APIs
Posted by za...@apache.org.
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 4f61d8585dbb26fa9823e25c04421bd097207e15
Author: Zakelly <za...@gmail.com>
AuthorDate: Wed Feb 5 17:31:41 2025 +0800
[FLINK-36439][docs] Documents for State V2 APIs
---
.../datastream-v2/context_and_state_processing.md | 7 +
.../docs/dev/datastream/fault-tolerance/state.md | 2 +-
.../dev/datastream/fault-tolerance/state_v2.md} | 793 ++++++---------------
.../datastream-v2/context_and_state_processing.md | 7 +
.../docs/dev/datastream/fault-tolerance/state.md | 2 +-
.../fault-tolerance/{state.md => state_v2.md} | 792 ++++++--------------
6 files changed, 454 insertions(+), 1149 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md b/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md
index cd17fe26d86..4e95270fac0 100644
--- a/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md
+++ b/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md
@@ -281,4 +281,11 @@ The specific rules are shown in the following table:
flink will check the legality of state declaration in advance. If an illegal state is declared in `usesStates` method, exception will be thrown at job compile time.
+### State Access Methods
+
+Note that the datastream v2 is using the state v2 APIs. For more detail about the state access
+methods, please refer to the
+[Using Keyed State V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2#using-keyed-state-v2" >}}).
+
+
{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
index e4d8c3b9c13..4f269a545dc 100644
--- a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
@@ -1,6 +1,6 @@
---
title: "Working with State"
-weight: 2
+weight: 1
type: docs
aliases:
- /zh/dev/stream/state/state.html
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
similarity index 51%
copy from docs/content/docs/dev/datastream/fault-tolerance/state.md
copy to docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
index 2486212055e..6ef16802f80 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md
@@ -1,10 +1,10 @@
---
-title: "Working with State"
+title: "Working with State V2"
weight: 2
type: docs
aliases:
- - /dev/stream/state/state.html
- - /apis/streaming/state.html
+ - /dev/stream/state/state_v2.html
+ - /apis/streaming/state_v2.html
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -25,20 +25,26 @@ specific language governing permissions and limitations
under the License.
-->
-# Working with State
+# Working with State V2 (New APIs)
-In this section you will learn about the APIs that Flink provides for writing
+In this section you will learn about the new APIs that Flink provides for writing
stateful programs. Please take a look at [Stateful Stream
Processing]({{< ref "docs/concepts/stateful-stream-processing" >}})
to learn about the concepts behind stateful stream processing.
+The new state API is designed to be more flexible than the previous API. User can perform
+asynchronous state operations, thus making it more powerful and more efficient.
+The asynchronous state access is essential for the state backend to be able to handle
+large state sizes and to be able to spill to remote file systems when necessary.
+
## Keyed DataStream
If you want to use keyed state, you first need to specify a key on a
`DataStream` that should be used to partition the state (and also the records
in the stream themselves). You can specify a key using `keyBy(KeySelector)`
-in Java/Scala API or `key_by(KeySelector)` in Python API on a `DataStream`.
-This will yield a `KeyedStream`, which then allows operations that use keyed state.
+in Java API on a `DataStream`. This will yield a `KeyedStream`, which then allows operations
+that use keyed state. You should perform `enableAsyncState()` on the `KeyedStream` to enable
+asynchronous state operations.
A key selector function takes a single record as input and returns the key for
that record. The key can be of any type and **must** be derived from
@@ -52,85 +58,132 @@ grouping operator.
The following example shows a key selector function that simply returns the
field of an object:
-{{< tabs "54e3bde7-659a-4683-81fa-112a2c81036b" >}}
+{{< tabs "54e3bde7-659a-4683-81fa-312a2c81036b" >}}
{{< tab "Java" >}}
```java
// some ordinary POJO
-public class WC {
+public class WordCount {
public String word;
public int count;
public String getWord() { return word; }
}
-DataStream<WC> words = // [...]
-KeyedStream<WC> keyed = words
- .keyBy(WC::getWord);
-```
-
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-// some ordinary case class
-case class WC(word: String, count: Int)
-val words: DataStream[WC] = // [...]
-val keyed = words.keyBy( _.word )
-```
-{{< /tab >}}
-
-{{< tab "Python" >}}
-```python
-words = # type: DataStream[Row]
-keyed = words.key_by(lambda row: row[0])
+DataStream<WordCount> words = // [...]
+KeyedStream<WordCount> keyed = words
+ .keyBy(WordCount::getWord).enableAsyncState();
```
-{{< /tab >}}
{{< /tabs >}}
-#### Tuple Keys and Expression Keys
-
-Flink also has two alternative ways of defining keys: tuple keys and expression
-keys in the Java/Scala API(still not supported in the Python API). With this you can
-specify keys using tuple field indices or expressions
-for selecting fields of objects. We don't recommend using these today but you
-can refer to the Javadoc of DataStream to learn about them. Using a KeySelector
-function is strictly superior: with Java lambdas they are easy to use and they
-have potentially less overhead at runtime.
-
{{< top >}}
-## Using Keyed State
+## Using Keyed State V2
-The keyed state interfaces provides access to different types of state that are all scoped to
-the key of the current input element. This means that this type of state can only be used
-on a `KeyedStream`, which can be created via `stream.keyBy(…)` in Java/Scala API or `stream.key_by(…)` in Python API.
+Unlike the previous state API, the new state API is designed for asynchronous state access.
+Each type of state gives two versions of the API: synchronous and asynchronous. The synchronous
+API is a blocking API that waits for the state access to complete. The asynchronous API is
+non-blocking and returns a `StateFuture` that will be completed when the state access
+is done. After that a callback or following logic will be invoked (if any).
+The asynchronous API is more efficient and should be used whenever possible.
+It is highly **not** recommended to mixed synchronous and asynchronous state access in the same
+user function.
-Now, we will first look at the different types of state available and then we will see
-how they can be used in a program. The available state primitives are:
+The keyed state interfaces provide access to different types of state that are all scoped to
+the key of the current input element. This means that this type of state can only be used
+on a `KeyedStream`, which can be created via `stream.keyBy(…)` in Java. And then most importantly,
+the keyed stream needs to be enabled for asynchronous state access by calling `enableAsyncState()`.
+The new API set are only available on `KeyedStream` with `enableAsyncState()` invoked.
+
+Now, we will look at the different types of state available, and then we will see
+how they can be used in a program. Since the synchronous APIs are identical with the original APIs,
+We only focus on the asynchronous ones here.
+
+### The Return Values
+
+First of all, we should get familiar with the return value of those asynchronous state access methods.
+
+`StateFuture<T>` is a future that will be completed with the result of the state access.
+The return bype is T. It provides multiple methods to handle the result, listed as:
+* `StateFuture<Void> thenAccept(Consumer<T>)`: This method takes a `Consumer` that will be called with the result
+ when the state access is done. It returns a `StateFuture<Void>`, which will be finished when the
+ `Consumer` is done.
+* `StateFuture<R> thenApply(Function<T, R>)`: This method takes a `Function` that will be called with the result
+ when the state access is done. The return value of the function will be the result of the following
+ `StateFuture`, which will be finished when the `Function` is done.
+* `StateFuture<R> thenCompose(Function<T, StateFuture<R>>)`: This method takes a `Function` that will
+ be called with the result when the state access is done. The return value of the function should be
+ a `StateFuture<R>`, which is exposed to the invoker of `thenCompose` as a return value.
+ The `StateFuture<R>` will be finished when the inner `StateFuture<R>` of `Function` is finished.
+* `StateFuture<R> thenCombine(StateFuture<U>, BiFunction<T, U, R>)`: This method takes another `StateFuture` and a
+ `BiFunction` that will be called with the results of both `StateFuture`s when they are done. The return
+ value of the `BiFunction` will be the result of the following `StateFuture`, which will be finished when
+ the `BiFunction` is done.
+
+Those methods are similar to the corresponding ones of the `CompletableFuture`. Besides these methods,
+keep in mind that `StateFuture` does not provide a `get()` method to block the current thread until the
+state access is done. This is because blocking the current thread may cause recursive blocking. The
+`StateFuture<T>` also provides conditional version of `thenAccept`, `thenApply`, `thenCompose` and `thenCombine`,
+which is for the case that the state access is done and the following logic will be split into two branches
+based on the result of the state access. The conditional version of those methods are `thenConditionallyAccept`,
+`thenConditionallyApply`, `thenConditionallyCompose` and `thenConditionallyCombine`.
+
+`StateIterator<T>` is an iterator that can be used to iterate over the elements of a state. It provides
+the following methods:
+* `boolean isEmpty()` : A synchronous method returns a `true` if the iterator has no elements, and `false` otherwise.
+* `StateFuture<Void> onNext(Consumer<T>)` : This method takes a `Consumer` that will be called with the next element
+ when the state access is done. It returns a `StateFuture<Void>`, which will be finished when the `Consumer` is done.
+ Also, a function version of `onNext` is provided, which is `StateFuture<Collection<R>> onNext(Function<T, R>)`. This method
+ takes a `Function` that will be called with the next element when the state access is done. The return value of the function
+ will be collected and returned as a collection of the following `StateFuture`, which will be
+ finished when the `Function` is done.
+
+We also provide a `StateFutureUtils` class that contains some utility methods to handle `StateFuture`s.
+These methods are:
+* `StateFuture<T> completedFuture(T)`: This method returns a completed `StateFuture` with the given value. This is
+useful when you want to return a constant value in a `thenCompose` method for further processing.
+* `StateFuture<Void> completedVoidFuture()`: This method returns a completed `StateFuture` with `null` value.
+A void value version of `completedFuture`.
+* `StateFuture<Collection<T>> combineAll(Collection<StateFuture<T>>)` : This method takes a collection of `StateFuture`s
+and returns a `StateFuture` that will be completed when all the input `StateFuture`s are completed. The result of the
+returned `StateFuture` is a collection of the results of the input `StateFuture`s. This is useful when you want to
+combine the results of multiple `StateFuture`s.
+* `StateFuture<Iterable<T>> toIterable(StateFuture<StateIterator<T>>)` : This method takes a `StateFuture` of `StateIterator`
+and returns a `StateFuture` of `Iterable`. The result of the returned `StateFuture` is an `Iterable` that contains all
+the elements of the `StateIterator`. This is useful when you want to convert a `StateIterator` to an `Iterable`.
+There is no good reason to do so, since this may disable the capability of lazy loading. Only useful when the further
+calculation depends on the whole data from the iterator.
+
+
+### State Primitives
+
+The available state primitives are:
* `ValueState<T>`: This keeps a value that can be updated and
retrieved (scoped to key of the input element as mentioned above, so there will possibly be one value
-for each key that the operation sees). The value can be set using `update(T)` and retrieved using
-`T value()`.
+for each key that the operation sees). The value can be set using `asyncUpdate(T)` and retrieved using
+`StateFuture<T> asyncValue()`.
-* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
-over all currently stored elements. Elements are added using `add(T)` or `addAll(List<T>)`, the Iterable can
-be retrieved using `Iterable<T> get()`. You can also override the existing list with `update(List<T>)`
+* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `StateIterator`
+over all currently stored elements. Elements are added using `asyncAdd(T)` or `asyncAddAll(List<T>)`,
+the Iterable can be retrieved using `StateFuture<StateIterator<T>> asyncGet()`.
+You can also override the existing list with `asyncUpdate(List<T>)`
* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
added to the state. The interface is similar to `ListState` but elements added using
-`add(T)` are reduced to an aggregate using a specified `ReduceFunction`.
+`asyncAdd(T)` are reduced to an aggregate using a specified `ReduceFunction`.
* `AggregatingState<IN, OUT>`: This keeps a single value that represents the aggregation of all values
added to the state. Contrary to `ReducingState`, the aggregate type may be different from the type
of elements that are added to the state. The interface is the same as for `ListState` but elements
-added using `add(IN)` are aggregated using a specified `AggregateFunction`.
+added using `asyncAdd(IN)` are aggregated using a specified `AggregateFunction`.
* `MapState<UK, UV>`: This keeps a list of mappings. You can put key-value pairs into the state and
-retrieve an `Iterable` over all currently stored mappings. Mappings are added using `put(UK, UV)` or
-`putAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `get(UK)`. The iterable
-views for mappings, keys and values can be retrieved using `entries()`, `keys()` and `values()` respectively.
-You can also use `isEmpty()` to check whether this map contains any key-value mappings.
+retrieve an `StateIterator` over all currently stored mappings. Mappings are added using `asyncPut(UK, UV)` or
+`asyncPutAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `asyncGet(UK)`.
+The iterable views for mappings, keys and values can be retrieved using `asyncEntries()`,
+`asyncKeys()` and `asyncValues()` respectively. You can also use `asyncIsEmpty()` to check whether
+this map contains any key-value mappings.
-All types of state also have a method `clear()` that clears the state for the currently
+All types of state also have a method `asyncClear()` that clears the state for the currently
active key, i.e. the key of the input element.
It is important to keep in mind that these state objects are only used for interfacing
@@ -145,6 +198,8 @@ that you can reference them), the type of the values that the state holds, and p
a user-specified function, such as a `ReduceFunction`. Depending on what type of state you
want to retrieve, you create either a `ValueStateDescriptor`, a `ListStateDescriptor`,
an `AggregatingStateDescriptor`, a `ReducingStateDescriptor`, or a `MapStateDescriptor`.
+To differentiate between the previous state APIs, you should use the `StateDescriptor`s under the
+`org.apache.flink.api.common.state.v2` package (note the **v2**).
State is accessed using the `RuntimeContext`, so it is only possible in *rich functions*.
Please see [here]({{< ref "docs/dev/datastream/user_defined_functions" >}}#rich-functions) for
@@ -159,7 +214,7 @@ is available in a `RichFunction` has these methods for accessing state:
This is an example `FlatMapFunction` that shows how all of the parts fit together:
-{{< tabs "348fd48c-fa36-4d27-9e53-9bffb74d8a86" >}}
+{{< tabs "348fd48c-fa36-4d27-9e53-9bffb74d8a87" >}}
{{< tab "Java" >}}
```java
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
@@ -173,22 +228,26 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
// access the state value
- Tuple2<Long, Long> currentSum = sum.value();
-
- // update the count
- currentSum.f0 += 1;
-
- // add the second field of the input value
- currentSum.f1 += input.f1;
-
- // update the state
- sum.update(currentSum);
-
- // if the count reaches 2, emit the average and clear the state
- if (currentSum.f0 >= 2) {
- out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
- sum.clear();
- }
+ sum.asyncValue().thenApply(currentSum -> {
+ // if it hasn't been used before, it will be null
+ Tuple2<Long, Long> current = currentSum == null ? Tuple2.of(0L, 0L) : currentSum;
+
+ // update the count
+ current.f0 += 1;
+
+ // add the second field of the input value
+ current.f1 += input.f1;
+
+ return current;
+ }).thenAccept(r -> {
+ // if the count reaches 2, emit the average and clear the state
+ if (r.f0 >= 2) {
+ out.collect(Tuple2.of(input.f0, r.f1 / r.f0));
+ sum.asyncClear();
+ } else {
+ sum.asyncUpdate(r);
+ }
+ });
}
@Override
@@ -196,8 +255,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average", // the state name
- TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
- Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
+ TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})); // type information
sum = getRuntimeContext().getState(descriptor);
}
}
@@ -205,117 +263,13 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
.keyBy(value -> value.f0)
+ .enableAsyncState()
.flatMap(new CountWindowAverage())
.print();
// the printed output will be (1,4) and (1,5)
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class CountWindowAverage extends RichFlatMapFunction[(Long, Long), (Long, Long)] {
-
- private var sum: ValueState[(Long, Long)] = _
-
- override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {
-
- // access the state value
- val tmpCurrentSum = sum.value
-
- // If it hasn't been used before, it will be null
- val currentSum = if (tmpCurrentSum != null) {
- tmpCurrentSum
- } else {
- (0L, 0L)
- }
-
- // update the count
- val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
-
- // update the state
- sum.update(newSum)
-
- // if the count reaches 2, emit the average and clear the state
- if (newSum._1 >= 2) {
- out.collect((input._1, newSum._2 / newSum._1))
- sum.clear()
- }
- }
-
- override def open(parameters: Configuration): Unit = {
- sum = getRuntimeContext.getState(
- new ValueStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
- )
- }
-}
-
-
-object ExampleCountWindowAverage extends App {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- env.fromCollection(List(
- (1L, 3L),
- (1L, 5L),
- (1L, 7L),
- (1L, 4L),
- (1L, 2L)
- )).keyBy(_._1)
- .flatMap(new CountWindowAverage())
- .print()
- // the printed output will be (1,4) and (1,5)
-
- env.execute("ExampleKeyedState")
-}
-```
-{{< /tab >}}
-
-{{< tab "Python" >}}
-```python
-from pyflink.common.typeinfo import Types
-from pyflink.datastream import StreamExecutionEnvironment, FlatMapFunction, RuntimeContext
-from pyflink.datastream.state import ValueStateDescriptor
-
-class CountWindowAverage(FlatMapFunction):
-
- def __init__(self):
- self.sum = None
-
- def open(self, runtime_context: RuntimeContext):
- descriptor = ValueStateDescriptor(
- "average", # the state name
- Types.PICKLED_BYTE_ARRAY() # type information
- )
- self.sum = runtime_context.get_state(descriptor)
-
- def flat_map(self, value):
- # access the state value
- current_sum = self.sum.value()
- if current_sum is None:
- current_sum = (0, 0)
-
- # update the count
- current_sum = (current_sum[0] + 1, current_sum[1] + value[1])
-
- # update the state
- self.sum.update(current_sum)
-
- # if the count reaches 2, emit the average and clear the state
- if current_sum[0] >= 2:
- self.sum.clear()
- yield value[0], int(current_sum[1] / current_sum[0])
-
-
-env = StreamExecutionEnvironment.get_execution_environment()
-env.from_collection([(1, 3), (1, 5), (1, 7), (1, 4), (1, 2)]) \
- .key_by(lambda row: row[0]) \
- .flat_map(CountWindowAverage()) \
- .print()
-
-env.execute()
-
-# the printed output will be (1,4) and (1,5)
-```
-{{< /tab >}}
{{< /tabs >}}
This example implements a poor man's counting window. We key the tuples by the first field
@@ -324,6 +278,45 @@ a `ValueState`. Once the count reaches 2 it will emit the average and clear the
we start over from `0`. Note that this would keep a different state value for each different input
key if we had tuples with different values in the first field.
+### Execution Order
+
+The state access methods are executed asynchronously. This means that the state access methods
+will not block the current thread. With synchronous APIs, the state access methods will be executed in
+the order they are called. However, with asynchronous APIs, the state access methods will be executed
+out of order, especially for those invokes for different incoming elements. For the above example,
+if the `flatMap` function is invoked for two different incoming elements A and B, the state access
+methods for A and B will be executed. Firstly, `asyncGet` is executed for A, then `asyncGet` is
+allowed to execute for B. The finish order of the two `asyncGet` is not guaranteed. Thus the order
+of continuation of the two `StateFuture`s is not guaranteed. Thus invokes of `asyncClear` or
+`asyncUpdate`for A and B are not determined.
+
+Although the state access methods are executed out of order, this not mean that all the user code
+are run in parallel. The user code in the `processElement`, `flatMap` or `thenXXxx` methods
+following the state access methods will be executed in a single thread (the task thread). So there
+is no concurrency issue for the user code.
+
+Typically, you don't need to worry about the execution order of the state access methods, but there
+is still some rules the Flink will ensure:
+* The execution order of user code entry `flatMap` for same-key elements are invoked strictly in
+order of element arrival.
+* The consumers or functions passed to the `thenXXxx` methods are executed in the order they are
+chained. If they are not chained, or there are multiple chains, the order is not guaranteed.
+
+### Best practice of asynchronous APIs
+
+The asynchronous APIs are designed to be more efficient and more powerful than the synchronous APIs.
+There are some best practices that you should follow when using the asynchronous APIs:
+* **Avoid mixing synchronous and asynchronous state access**
+* Use chaining of `thenXXxx` methods to handle the result of the state access and then gives another
+ state access or result. Divide the logic into multiple steps split by `thenXXxx` methods.
+* Avoid accessing mutable members of the user function (`RichFlatMapFunction`). Since the state access
+ methods are executed out of order, the mutable members may be accessed in unpredictable order.
+ Instead, use the result of the state access to pass the data between different steps. The
+ `StateFutureUtils.completedFuture` or `thenApply` method can be used to pass the data. Or use
+ a captured container (`AtomicReference`) which is initialized for each invoke of the `flatMap`
+ to share between lambdas.
+
+
### State Time-To-Live (TTL)
A *time-to-live* (TTL) can be assigned to the keyed state of any type. If a TTL is configured and a
@@ -336,7 +329,7 @@ expire independently.
In order to use state TTL one must first build a `StateTtlConfig` configuration object. The TTL
functionality can then be enabled in any state descriptor by passing the configuration:
-{{< tabs "43221f5b-b56c-42e5-833e-ed416e9d9b4e" >}}
+{{< tabs "43221f0b-b56c-42e5-833e-ed416e5f9b4e" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -353,38 +346,6 @@ ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text
stateDescriptor.enableTimeToLive(ttlConfig);
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-import org.apache.flink.api.common.state.ValueStateDescriptor
-import java.time.Duration
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
- .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
- .build
-
-val stateDescriptor = new ValueStateDescriptor[String]("text state", classOf[String])
-stateDescriptor.enableTimeToLive(ttlConfig)
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.common.typeinfo import Types
-from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite) \
- .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired) \
- .build()
-
-state_descriptor = ValueStateDescriptor("text state", Types.STRING())
-state_descriptor.enable_time_to_live(ttl_config)
-```
-{{< /tab >}}
{{< /tabs >}}
The configuration has several options to consider:
@@ -420,14 +381,15 @@ Another option `ReturnExpiredIfNotCleanedUp` allows to return the expired state
- The state backends store the timestamp of the last modification along with the user value,
which means that enabling this feature increases consumption of state storage.
Heap state backend stores an additional Java object with a reference to the user state object
-and a primitive long value in memory. The RocksDB state backend adds 8 bytes per stored value, list entry or map entry.
+and a primitive long value in memory. The RocksDB/ForSt state backend adds 8 bytes per stored value,
+list entry or map entry.
- Only TTLs in reference to *processing time* are currently supported.
- Trying to restore state, which was previously configured without TTL, using TTL enabled descriptor or vice versa
will lead to compatibility failure and `StateMigrationException`.
-- The TTL configuration is not part of check- or savepoints but rather a way of how Flink treats it in the currently running job.
+- The TTL configuration is not part of checkpoints or savepoints but rather a way of how Flink treats it in the currently running job.
- It is not recommended to restore checkpoint state with adjusting the ttl from a short value to a long value,
which may cause potential data errors.
@@ -442,7 +404,7 @@ If the serializer does not support null values, it can be wrapped with `Nullable
By default, expired values are explicitly removed on read, such as `ValueState#value`, and periodically garbage collected
in the background if supported by the configured state backend. Background cleanup can be disabled in the `StateTtlConfig`:
-{{< tabs "dfad4bf4-84ec-49a9-8372-c63b59122603" >}}
+{{< tabs "df774bf4-84ec-49a9-8372-c63b59122608" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -452,149 +414,25 @@ StateTtlConfig ttlConfig = StateTtlConfig
.build();
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .disableCleanupInBackground
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .disable_cleanup_in_background() \
- .build()
-```
-{{< /tab >}}
{{< /tabs >}}
-For more fine-grained control over some special cleanup in background, you can configure it separately as described below.
-Currently, heap state backend relies on incremental cleanup and RocksDB backend uses compaction filter for background cleanup.
-
-##### Cleanup in full snapshot
+{{ <hint info> }}
+Currently the ForSt State Backend only cleanup the state in compaction process. If you are using
+a state backend other than ForSt and want to read about other cleanup strategies, please refer to the
+[State V1 documentation]({{< ref "docs/dev/datastream/fault-tolerance/state#cleanup-of-expired-state" >}})
+about the expired state cleanup.
+{{ </hint > }
-Additionally, you can activate the cleanup at the moment of taking the full state snapshot which
-will reduce its size. The local state is not cleaned up under the current implementation
-but it will not include the removed expired state in case of restoration from the previous snapshot.
-It can be configured in `StateTtlConfig`:
-
-{{< tabs "ad8a4c67-222c-4beb-9547-ca5f8085695c" >}}
-{{< tab "Java" >}}
-```java
-import org.apache.flink.api.common.state.StateTtlConfig;
-import java.time.Duration;
-
-StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupFullSnapshot()
- .build();
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-import java.time.Duration
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupFullSnapshot
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_full_snapshot() \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-This option is not applicable for the incremental checkpointing in the RocksDB state backend.
-
-{{< hint info >}}
-For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
-{{< /hint >}}
-
-##### Incremental cleanup
-
-Another option is to trigger cleanup of some state entries incrementally.
-The trigger can be a callback from each state access or/and each record processing.
-If this cleanup strategy is active for certain state,
-The storage backend keeps a lazy global iterator for this state over all its entries.
-Every time incremental cleanup is triggered, the iterator is advanced.
-The traversed state entries are checked and expired ones are cleaned up.
-
-This feature can be configured in `StateTtlConfig`:
+##### Cleanup during compaction
-{{< tabs "e8c96bb9-6485-4f86-8fdb-5e538e538827" >}}
-{{< tab "Java" >}}
-```java
-import org.apache.flink.api.common.state.StateTtlConfig;
- StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupIncrementally(10, true)
- .build();
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupIncrementally(10, true)
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_incrementally(10, True) \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-This strategy has two parameters. The first one is number of checked state entries per each cleanup triggering.
-It is always triggered per each state access.
-The second parameter defines whether to trigger cleanup additionally per each record processing.
-The default background cleanup for heap backend checks 5 entries without cleanup per record processing.
-
-**Notes:**
-- If no access happens to the state or no records are processed, expired state will persist.
-- Time spent for the incremental cleanup increases record processing latency.
-- At the moment incremental cleanup is implemented only for Heap state backend. Setting it for RocksDB will have no effect.
-- If heap state backend is used with synchronous snapshotting, the global iterator keeps a copy of all keys
-while iterating because of its specific implementation which does not support concurrent modifications.
-Enabling of this feature will increase memory consumption then. Asynchronous snapshotting does not have this problem.
-- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
-
-##### Cleanup during RocksDB compaction
-
-If the RocksDB state backend is used, a Flink specific compaction filter will be called for the background cleanup.
-RocksDB periodically runs asynchronous compactions to merge state updates and reduce storage.
+If the ForSt state backend is used, a Flink specific compaction filter will be called for the background cleanup.
+ForSt periodically runs asynchronous compactions to merge state updates and reduce storage.
Flink compaction filter checks expiration timestamp of state entries with TTL
-and excludes expired values.
+and excludes expired values.
This feature can be configured in `StateTtlConfig`:
-{{< tabs "4db505be-2c9c-4ef7-aa01-74763968b7a1" >}}
+{{< tabs "4db505be-2c9c-6ef7-aa01-74763968b7a9" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -605,40 +443,18 @@ StateTtlConfig ttlConfig = StateTtlConfig
.build();
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common import Duration
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_in_rocksdb_compact_filter(1000, Duration.of_hours(1)) \
- .build()
-```
-{{< /tab >}}
{{< /tabs >}}
-RocksDB compaction filter will query current timestamp, used to check expiration, from Flink every time
+ForSt compaction filter will query current timestamp, used to check expiration, from Flink every time
after processing certain number of state entries.
-You can change it and pass a custom value to
-`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
-Updating the timestamp more often can improve cleanup speed
+You can change it and pass a custom value to
+`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
+Updating the timestamp more often can improve cleanup speed
but it decreases compaction performance because it uses JNI call from native code.
-The default background cleanup for RocksDB backend queries the current timestamp each time 1000 entries have been processed.
+The default background cleanup for ForSt backend queries the current timestamp each time 1000 entries have been processed.
-Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed.
-Files older than this value will be picked up for compaction, and re-written to the same level as they were before.
+Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed.
+Files older than this value will be picked up for compaction, and re-written to the same level as they were before.
It makes sure a file goes through compaction filters periodically.
You can change it and pass a custom value to
`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Duration periodicCompactionTime)` method.
@@ -646,42 +462,22 @@ The default value of Periodic compaction seconds is 30 days.
You could set it to 0 to turn off periodic compaction or set a small value to speed up expired state entries cleanup, but it
would trigger more compactions.
-You can activate debug logs from the native code of RocksDB filter
+You can activate debug logs from the native code of ForSt filter
by activating debug level for `FlinkCompactionFilter`:
-`log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG`
+`log4j.logger.org.forstdb.FlinkCompactionFilter=DEBUG`
-**Notes:**
-- Calling of TTL filter during compaction slows it down.
-The TTL filter has to parse timestamp of last access and check its expiration
-for every stored state entry per key which is being compacted.
-In case of collection state type (list or map) the check is also invoked per stored element.
+**Notes:**
+- Calling of TTL filter during compaction slows it down.
+ The TTL filter has to parse timestamp of last access and check its expiration
+ for every stored state entry per key which is being compacted.
+ In case of collection state type (list or map) the check is also invoked per stored element.
- If this feature is used with a list state which has elements with non-fixed byte length,
-the native TTL filter has to call additionally a Flink java type serializer of the element over JNI per each state entry
-where at least the first element has expired to determine the offset of the next unexpired element.
-- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
+ the native TTL filter has to call additionally a Flink java type serializer of the element over JNI per each state entry
+ where at least the first element has expired to determine the offset of the next unexpired element.
+- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
+ e.g. after restart from savepoint.
- Periodic compaction could only work when TTL is enabled.
-
-### State in the Scala DataStream API
-
-In addition to the interface described above, the Scala API has shortcuts for stateful
-`map()` or `flatMap()` functions with a single `ValueState` on `KeyedStream`. The user function
-gets the current value of the `ValueState` in an `Option` and must return an updated value that
-will be used to update the state.
-
-```scala
-val stream: DataStream[(String, Int)] = ...
-
-val counts: DataStream[(String, Int)] = stream
- .keyBy(_._1)
- .mapWithState((in: (String, Int), count: Option[Int]) =>
- count match {
- case Some(c) => ( (in._1, c), Some(c + in._2) )
- case None => ( (in._1, 0), Some(in._2) )
- })
-```
-
## Operator State
*Operator State* (or *non-keyed state*) is state that is bound to one
@@ -697,8 +493,6 @@ In a typical stateful Flink Application you don't need operators state. It is
mostly a special type of state that is used in source/sink implementations and
scenarios where you don't have a key by which state can be partitioned.
-**Notes:** Operator state is still not supported in Python DataStream API.
-
## Broadcast State
*Broadcast State* is a special type of *Operator State*. It was introduced to
@@ -762,7 +556,7 @@ Below is an example of a stateful `SinkFunction` that uses `CheckpointedFunction
to buffer elements before sending them to the outside world. It demonstrates
the basic even-split redistribution list state:
-{{< tabs "5b99313c-1801-486c-a085-ce66882a2e49" >}}
+{{< tabs "5b99663c-1801-486c-a085-ce66882a2k49" >}}
{{< tab "Java" >}}
```java
public class BufferingSink
@@ -814,49 +608,6 @@ public class BufferingSink
}
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class BufferingSink(threshold: Int = 0)
- extends SinkFunction[(String, Int)]
- with CheckpointedFunction {
-
- @transient
- private var checkpointedState: ListState[(String, Int)] = _
-
- private val bufferedElements = ListBuffer[(String, Int)]()
-
- override def invoke(value: (String, Int), context: Context): Unit = {
- bufferedElements += value
- if (bufferedElements.size >= threshold) {
- for (element <- bufferedElements) {
- // send it to the sink
- }
- bufferedElements.clear()
- }
- }
-
- override def snapshotState(context: FunctionSnapshotContext): Unit = {
- checkpointedState.update(bufferedElements.asJava)
- }
-
- override def initializeState(context: FunctionInitializationContext): Unit = {
- val descriptor = new ListStateDescriptor[(String, Int)](
- "buffered-elements",
- TypeInformation.of(new TypeHint[(String, Int)]() {})
- )
-
- checkpointedState = context.getOperatorStateStore.getListState(descriptor)
-
- if(context.isRestored) {
- for(element <- checkpointedState.get().asScala) {
- bufferedElements += element
- }
- }
- }
-
-}
-```
-{{< /tab >}}
{{< /tabs >}}
The `initializeState` method takes as argument a `FunctionInitializationContext`. This is used to initialize
@@ -879,18 +630,6 @@ ListStateDescriptor<Tuple2<String, Integer>> descriptor =
checkpointedState = context.getOperatorStateStore().getListState(descriptor);
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-
-val descriptor = new ListStateDescriptor[(String, Long)](
- "buffered-elements",
- TypeInformation.of(new TypeHint[(String, Long)]() {})
-)
-
-checkpointedState = context.getOperatorStateStore.getListState(descriptor)
-
-```
{{< /tab >}}
{{< /tabs >}}
The naming convention of the state access methods contain its redistribution
@@ -909,110 +648,16 @@ of all objects included by the previous checkpoint, and is then filled with the
As a side note, the keyed state can also be initialized in the `initializeState()` method. This can be done
using the provided `FunctionInitializationContext`.
-### Stateful Source Functions
-
-Stateful sources require a bit more care as opposed to other operators.
-In order to make the updates to the state and output collection atomic (required for exactly-once semantics
-on failure/recovery), the user is required to get a lock from the source's context.
-
-{{< tabs "aa821e89-7cc6-4c1e-be5b-90bfa0df6036" >}}
-{{< tab "Java" >}}
-```java
-public static class CounterSource
- extends RichParallelSourceFunction<Long>
- implements CheckpointedFunction {
-
- /** current offset for exactly once semantics */
- private Long offset = 0L;
-
- /** flag for job cancellation */
- private volatile boolean isRunning = true;
-
- /** Our state object. */
- private ListState<Long> state;
-
- @Override
- public void run(SourceContext<Long> ctx) {
- final Object lock = ctx.getCheckpointLock();
-
- while (isRunning) {
- // output and state update are atomic
- synchronized (lock) {
- ctx.collect(offset);
- offset += 1;
- }
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
- state = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
- "state",
- LongSerializer.INSTANCE));
-
- // restore any state that we might already have to our fields, initialize state
- // is also called in case of restore.
- for (Long l : state.get()) {
- offset = l;
- }
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- state.update(Collections.singletonList(offset));
- }
-}
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class CounterSource
- extends RichParallelSourceFunction[Long]
- with CheckpointedFunction {
-
- @volatile
- private var isRunning = true
-
- private var offset = 0L
- private var state: ListState[Long] = _
-
- override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
- val lock = ctx.getCheckpointLock
-
- while (isRunning) {
- // output and state update are atomic
- lock.synchronized({
- ctx.collect(offset)
-
- offset += 1
- })
- }
- }
+{{< top >}}
- override def cancel(): Unit = isRunning = false
-
- override def initializeState(context: FunctionInitializationContext): Unit = {
- state = context.getOperatorStateStore.getListState(
- new ListStateDescriptor[Long]("state", classOf[Long]))
+## Migrate from the Old State API
- for (l <- state.get().asScala) {
- offset = l
- }
- }
+It is very easy to migrate from the old state API to the new one. Please take the following steps:
- override def snapshotState(context: FunctionSnapshotContext): Unit = {
- state.update(java.util.Collections.singletonList(offset))
- }
-}
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-Some operators might need the information when a checkpoint is fully acknowledged by Flink to communicate that with the outside world. In this case see the `org.apache.flink.api.common.state.CheckpointListener` interface.
-
-{{< top >}}
+1. Invoke `enableAsyncState()` on the `KeyedStream` to enable the new state API.
+2. Replace the `StateDescriptor`s with the new ones under `v2` package. Also, replace the old state
+handles with the new ones under `v2` package.
+3. Rewrite the old state access methods with the new asynchronous ones.
+4. It is recommended to use ForSt State Backend for the new state API, which could perform
+async state access. Other state backends only support synchronous execution of state access,
+although it can be used with the new state API.
diff --git a/docs/content/docs/dev/datastream-v2/context_and_state_processing.md b/docs/content/docs/dev/datastream-v2/context_and_state_processing.md
index cd17fe26d86..4e95270fac0 100644
--- a/docs/content/docs/dev/datastream-v2/context_and_state_processing.md
+++ b/docs/content/docs/dev/datastream-v2/context_and_state_processing.md
@@ -281,4 +281,11 @@ The specific rules are shown in the following table:
flink will check the legality of state declaration in advance. If an illegal state is declared in `usesStates` method, exception will be thrown at job compile time.
+### State Access Methods
+
+Note that the datastream v2 is using the state v2 APIs. For more detail about the state access
+methods, please refer to the
+[Using Keyed State V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2#using-keyed-state-v2" >}}).
+
+
{{< top >}}
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content/docs/dev/datastream/fault-tolerance/state.md
index 2486212055e..86714d0b7cb 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/state.md
@@ -1,6 +1,6 @@
---
title: "Working with State"
-weight: 2
+weight: 1
type: docs
aliases:
- /dev/stream/state/state.html
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
similarity index 51%
copy from docs/content/docs/dev/datastream/fault-tolerance/state.md
copy to docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
index 2486212055e..d1abd47ae49 100644
--- a/docs/content/docs/dev/datastream/fault-tolerance/state.md
+++ b/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md
@@ -1,10 +1,10 @@
---
-title: "Working with State"
+title: "Working with State V2"
weight: 2
type: docs
aliases:
- - /dev/stream/state/state.html
- - /apis/streaming/state.html
+ - /dev/stream/state/state_v2.html
+ - /apis/streaming/state_v2.html
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -25,20 +25,26 @@ specific language governing permissions and limitations
under the License.
-->
-# Working with State
+# Working with State V2 (New APIs)
-In this section you will learn about the APIs that Flink provides for writing
+In this section you will learn about the new APIs that Flink provides for writing
stateful programs. Please take a look at [Stateful Stream
Processing]({{< ref "docs/concepts/stateful-stream-processing" >}})
to learn about the concepts behind stateful stream processing.
+The new state API is designed to be more flexible than the previous API. User can perform
+asynchronous state operations, thus making it more powerful and more efficient.
+The asynchronous state access is essential for the state backend to be able to handle
+large state sizes and to be able to spill to remote file systems when necessary.
+
## Keyed DataStream
If you want to use keyed state, you first need to specify a key on a
`DataStream` that should be used to partition the state (and also the records
in the stream themselves). You can specify a key using `keyBy(KeySelector)`
-in Java/Scala API or `key_by(KeySelector)` in Python API on a `DataStream`.
-This will yield a `KeyedStream`, which then allows operations that use keyed state.
+in Java API on a `DataStream`. This will yield a `KeyedStream`, which then allows operations
+that use keyed state. You should perform `enableAsyncState()` on the `KeyedStream` to enable
+asynchronous state operations.
A key selector function takes a single record as input and returns the key for
that record. The key can be of any type and **must** be derived from
@@ -52,85 +58,132 @@ grouping operator.
The following example shows a key selector function that simply returns the
field of an object:
-{{< tabs "54e3bde7-659a-4683-81fa-112a2c81036b" >}}
+{{< tabs "54e3bde7-659a-4683-81fa-312a2c81036b" >}}
{{< tab "Java" >}}
```java
// some ordinary POJO
-public class WC {
+public class WordCount {
public String word;
public int count;
public String getWord() { return word; }
}
-DataStream<WC> words = // [...]
-KeyedStream<WC> keyed = words
- .keyBy(WC::getWord);
-```
-
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-// some ordinary case class
-case class WC(word: String, count: Int)
-val words: DataStream[WC] = // [...]
-val keyed = words.keyBy( _.word )
-```
-{{< /tab >}}
-
-{{< tab "Python" >}}
-```python
-words = # type: DataStream[Row]
-keyed = words.key_by(lambda row: row[0])
+DataStream<WordCount> words = // [...]
+KeyedStream<WordCount> keyed = words
+ .keyBy(WordCount::getWord).enableAsyncState();
```
-{{< /tab >}}
{{< /tabs >}}
-#### Tuple Keys and Expression Keys
-
-Flink also has two alternative ways of defining keys: tuple keys and expression
-keys in the Java/Scala API(still not supported in the Python API). With this you can
-specify keys using tuple field indices or expressions
-for selecting fields of objects. We don't recommend using these today but you
-can refer to the Javadoc of DataStream to learn about them. Using a KeySelector
-function is strictly superior: with Java lambdas they are easy to use and they
-have potentially less overhead at runtime.
-
{{< top >}}
-## Using Keyed State
+## Using Keyed State V2
-The keyed state interfaces provides access to different types of state that are all scoped to
-the key of the current input element. This means that this type of state can only be used
-on a `KeyedStream`, which can be created via `stream.keyBy(…)` in Java/Scala API or `stream.key_by(…)` in Python API.
+Unlike the previous state API, the new state API is designed for asynchronous state access.
+Each type of state gives two versions of the API: synchronous and asynchronous. The synchronous
+API is a blocking API that waits for the state access to complete. The asynchronous API is
+non-blocking and returns a `StateFuture` that will be completed when the state access
+is done. After that a callback or following logic will be invoked (if any).
+The asynchronous API is more efficient and should be used whenever possible.
+It is highly **not** recommended to mixed synchronous and asynchronous state access in the same
+user function.
-Now, we will first look at the different types of state available and then we will see
-how they can be used in a program. The available state primitives are:
+The keyed state interfaces provide access to different types of state that are all scoped to
+the key of the current input element. This means that this type of state can only be used
+on a `KeyedStream`, which can be created via `stream.keyBy(…)` in Java. And then most importantly,
+the keyed stream needs to be enabled for asynchronous state access by calling `enableAsyncState()`.
+The new API set are only available on `KeyedStream` with `enableAsyncState()` invoked.
+
+Now, we will look at the different types of state available, and then we will see
+how they can be used in a program. Since the synchronous APIs are identical with the original APIs,
+We only focus on the asynchronous ones here.
+
+### The Return Values
+
+First of all, we should get familiar with the return value of those asynchronous state access methods.
+
+`StateFuture<T>` is a future that will be completed with the result of the state access.
+The return bype is T. It provides multiple methods to handle the result, listed as:
+* `StateFuture<Void> thenAccept(Consumer<T>)`: This method takes a `Consumer` that will be called with the result
+ when the state access is done. It returns a `StateFuture<Void>`, which will be finished when the
+ `Consumer` is done.
+* `StateFuture<R> thenApply(Function<T, R>)`: This method takes a `Function` that will be called with the result
+ when the state access is done. The return value of the function will be the result of the following
+ `StateFuture`, which will be finished when the `Function` is done.
+* `StateFuture<R> thenCompose(Function<T, StateFuture<R>>)`: This method takes a `Function` that will
+ be called with the result when the state access is done. The return value of the function should be
+ a `StateFuture<R>`, which is exposed to the invoker of `thenCompose` as a return value.
+ The `StateFuture<R>` will be finished when the inner `StateFuture<R>` of `Function` is finished.
+* `StateFuture<R> thenCombine(StateFuture<U>, BiFunction<T, U, R>)`: This method takes another `StateFuture` and a
+ `BiFunction` that will be called with the results of both `StateFuture`s when they are done. The return
+ value of the `BiFunction` will be the result of the following `StateFuture`, which will be finished when
+ the `BiFunction` is done.
+
+Those methods are similar to the corresponding ones of the `CompletableFuture`. Besides these methods,
+keep in mind that `StateFuture` does not provide a `get()` method to block the current thread until the
+state access is done. This is because blocking the current thread may cause recursive blocking. The
+`StateFuture<T>` also provides conditional version of `thenAccept`, `thenApply`, `thenCompose` and `thenCombine`,
+which is for the case that the state access is done and the following logic will be split into two branches
+based on the result of the state access. The conditional version of those methods are `thenConditionallyAccept`,
+`thenConditionallyApply`, `thenConditionallyCompose` and `thenConditionallyCombine`.
+
+`StateIterator<T>` is an iterator that can be used to iterate over the elements of a state. It provides
+the following methods:
+* `boolean isEmpty()` : A synchronous method returns a `true` if the iterator has no elements, and `false` otherwise.
+* `StateFuture<Void> onNext(Consumer<T>)` : This method takes a `Consumer` that will be called with the next element
+ when the state access is done. It returns a `StateFuture<Void>`, which will be finished when the `Consumer` is done.
+ Also, a function version of `onNext` is provided, which is `StateFuture<Collection<R>> onNext(Function<T, R>)`. This method
+ takes a `Function` that will be called with the next element when the state access is done. The return value of the function
+ will be collected and returned as a collection of the following `StateFuture`, which will be
+ finished when the `Function` is done.
+
+We also provide a `StateFutureUtils` class that contains some utility methods to handle `StateFuture`s.
+These methods are:
+* `StateFuture<T> completedFuture(T)`: This method returns a completed `StateFuture` with the given value. This is
+useful when you want to return a constant value in a `thenCompose` method for further processing.
+* `StateFuture<Void> completedVoidFuture()`: This method returns a completed `StateFuture` with `null` value.
+A void value version of `completedFuture`.
+* `StateFuture<Collection<T>> combineAll(Collection<StateFuture<T>>)` : This method takes a collection of `StateFuture`s
+and returns a `StateFuture` that will be completed when all the input `StateFuture`s are completed. The result of the
+returned `StateFuture` is a collection of the results of the input `StateFuture`s. This is useful when you want to
+combine the results of multiple `StateFuture`s.
+* `StateFuture<Iterable<T>> toIterable(StateFuture<StateIterator<T>>)` : This method takes a `StateFuture` of `StateIterator`
+and returns a `StateFuture` of `Iterable`. The result of the returned `StateFuture` is an `Iterable` that contains all
+the elements of the `StateIterator`. This is useful when you want to convert a `StateIterator` to an `Iterable`.
+There is no good reason to do so, since this may disable the capability of lazy loading. Only useful when the further
+calculation depends on the whole data from the iterator.
+
+
+### State Primitives
+
+The available state primitives are:
* `ValueState<T>`: This keeps a value that can be updated and
retrieved (scoped to key of the input element as mentioned above, so there will possibly be one value
-for each key that the operation sees). The value can be set using `update(T)` and retrieved using
-`T value()`.
+for each key that the operation sees). The value can be set using `asyncUpdate(T)` and retrieved using
+`StateFuture<T> asyncValue()`.
-* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
-over all currently stored elements. Elements are added using `add(T)` or `addAll(List<T>)`, the Iterable can
-be retrieved using `Iterable<T> get()`. You can also override the existing list with `update(List<T>)`
+* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `StateIterator`
+over all currently stored elements. Elements are added using `asyncAdd(T)` or `asyncAddAll(List<T>)`,
+the Iterable can be retrieved using `StateFuture<StateIterator<T>> asyncGet()`.
+You can also override the existing list with `asyncUpdate(List<T>)`
* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
added to the state. The interface is similar to `ListState` but elements added using
-`add(T)` are reduced to an aggregate using a specified `ReduceFunction`.
+`asyncAdd(T)` are reduced to an aggregate using a specified `ReduceFunction`.
* `AggregatingState<IN, OUT>`: This keeps a single value that represents the aggregation of all values
added to the state. Contrary to `ReducingState`, the aggregate type may be different from the type
of elements that are added to the state. The interface is the same as for `ListState` but elements
-added using `add(IN)` are aggregated using a specified `AggregateFunction`.
+added using `asyncAdd(IN)` are aggregated using a specified `AggregateFunction`.
* `MapState<UK, UV>`: This keeps a list of mappings. You can put key-value pairs into the state and
-retrieve an `Iterable` over all currently stored mappings. Mappings are added using `put(UK, UV)` or
-`putAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `get(UK)`. The iterable
-views for mappings, keys and values can be retrieved using `entries()`, `keys()` and `values()` respectively.
-You can also use `isEmpty()` to check whether this map contains any key-value mappings.
+retrieve an `StateIterator` over all currently stored mappings. Mappings are added using `asyncPut(UK, UV)` or
+`asyncPutAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `asyncGet(UK)`.
+The iterable views for mappings, keys and values can be retrieved using `asyncEntries()`,
+`asyncKeys()` and `asyncValues()` respectively. You can also use `asyncIsEmpty()` to check whether
+this map contains any key-value mappings.
-All types of state also have a method `clear()` that clears the state for the currently
+All types of state also have a method `asyncClear()` that clears the state for the currently
active key, i.e. the key of the input element.
It is important to keep in mind that these state objects are only used for interfacing
@@ -145,6 +198,8 @@ that you can reference them), the type of the values that the state holds, and p
a user-specified function, such as a `ReduceFunction`. Depending on what type of state you
want to retrieve, you create either a `ValueStateDescriptor`, a `ListStateDescriptor`,
an `AggregatingStateDescriptor`, a `ReducingStateDescriptor`, or a `MapStateDescriptor`.
+To differentiate between the previous state APIs, you should use the `StateDescriptor`s under the
+`org.apache.flink.api.common.state.v2` package (note the **v2**).
State is accessed using the `RuntimeContext`, so it is only possible in *rich functions*.
Please see [here]({{< ref "docs/dev/datastream/user_defined_functions" >}}#rich-functions) for
@@ -159,7 +214,7 @@ is available in a `RichFunction` has these methods for accessing state:
This is an example `FlatMapFunction` that shows how all of the parts fit together:
-{{< tabs "348fd48c-fa36-4d27-9e53-9bffb74d8a86" >}}
+{{< tabs "348fd48c-fa36-4d27-9e53-9bffb74d8a87" >}}
{{< tab "Java" >}}
```java
public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
@@ -173,22 +228,26 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
// access the state value
- Tuple2<Long, Long> currentSum = sum.value();
-
- // update the count
- currentSum.f0 += 1;
-
- // add the second field of the input value
- currentSum.f1 += input.f1;
-
- // update the state
- sum.update(currentSum);
-
- // if the count reaches 2, emit the average and clear the state
- if (currentSum.f0 >= 2) {
- out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
- sum.clear();
- }
+ sum.asyncValue().thenApply(currentSum -> {
+ // if it hasn't been used before, it will be null
+ Tuple2<Long, Long> current = currentSum == null ? Tuple2.of(0L, 0L) : currentSum;
+
+ // update the count
+ current.f0 += 1;
+
+ // add the second field of the input value
+ current.f1 += input.f1;
+
+ return current;
+ }).thenAccept(r -> {
+ // if the count reaches 2, emit the average and clear the state
+ if (r.f0 >= 2) {
+ out.collect(Tuple2.of(input.f0, r.f1 / r.f0));
+ sum.asyncClear();
+ } else {
+ sum.asyncUpdate(r);
+ }
+ });
}
@Override
@@ -196,8 +255,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
"average", // the state name
- TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
- Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
+ TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})); // type information
sum = getRuntimeContext().getState(descriptor);
}
}
@@ -205,117 +263,13 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
.keyBy(value -> value.f0)
+ .enableAsyncState()
.flatMap(new CountWindowAverage())
.print();
// the printed output will be (1,4) and (1,5)
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class CountWindowAverage extends RichFlatMapFunction[(Long, Long), (Long, Long)] {
-
- private var sum: ValueState[(Long, Long)] = _
-
- override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {
-
- // access the state value
- val tmpCurrentSum = sum.value
-
- // If it hasn't been used before, it will be null
- val currentSum = if (tmpCurrentSum != null) {
- tmpCurrentSum
- } else {
- (0L, 0L)
- }
-
- // update the count
- val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
-
- // update the state
- sum.update(newSum)
-
- // if the count reaches 2, emit the average and clear the state
- if (newSum._1 >= 2) {
- out.collect((input._1, newSum._2 / newSum._1))
- sum.clear()
- }
- }
-
- override def open(parameters: Configuration): Unit = {
- sum = getRuntimeContext.getState(
- new ValueStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
- )
- }
-}
-
-
-object ExampleCountWindowAverage extends App {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- env.fromCollection(List(
- (1L, 3L),
- (1L, 5L),
- (1L, 7L),
- (1L, 4L),
- (1L, 2L)
- )).keyBy(_._1)
- .flatMap(new CountWindowAverage())
- .print()
- // the printed output will be (1,4) and (1,5)
-
- env.execute("ExampleKeyedState")
-}
-```
-{{< /tab >}}
-
-{{< tab "Python" >}}
-```python
-from pyflink.common.typeinfo import Types
-from pyflink.datastream import StreamExecutionEnvironment, FlatMapFunction, RuntimeContext
-from pyflink.datastream.state import ValueStateDescriptor
-
-class CountWindowAverage(FlatMapFunction):
-
- def __init__(self):
- self.sum = None
-
- def open(self, runtime_context: RuntimeContext):
- descriptor = ValueStateDescriptor(
- "average", # the state name
- Types.PICKLED_BYTE_ARRAY() # type information
- )
- self.sum = runtime_context.get_state(descriptor)
-
- def flat_map(self, value):
- # access the state value
- current_sum = self.sum.value()
- if current_sum is None:
- current_sum = (0, 0)
-
- # update the count
- current_sum = (current_sum[0] + 1, current_sum[1] + value[1])
-
- # update the state
- self.sum.update(current_sum)
-
- # if the count reaches 2, emit the average and clear the state
- if current_sum[0] >= 2:
- self.sum.clear()
- yield value[0], int(current_sum[1] / current_sum[0])
-
-
-env = StreamExecutionEnvironment.get_execution_environment()
-env.from_collection([(1, 3), (1, 5), (1, 7), (1, 4), (1, 2)]) \
- .key_by(lambda row: row[0]) \
- .flat_map(CountWindowAverage()) \
- .print()
-
-env.execute()
-
-# the printed output will be (1,4) and (1,5)
-```
-{{< /tab >}}
{{< /tabs >}}
This example implements a poor man's counting window. We key the tuples by the first field
@@ -324,6 +278,45 @@ a `ValueState`. Once the count reaches 2 it will emit the average and clear the
we start over from `0`. Note that this would keep a different state value for each different input
key if we had tuples with different values in the first field.
+### Execution Order
+
+The state access methods are executed asynchronously. This means that the state access methods
+will not block the current thread. With synchronous APIs, the state access methods will be executed in
+the order they are called. However, with asynchronous APIs, the state access methods will be executed
+out of order, especially for those invokes for different incoming elements. For the above example,
+if the `flatMap` function is invoked for two different incoming elements A and B, the state access
+methods for A and B will be executed. Firstly, `asyncGet` is executed for A, then `asyncGet` is
+allowed to execute for B. The finish order of the two `asyncGet` is not guaranteed. Thus the order
+of continuation of the two `StateFuture`s is not guaranteed. Thus invokes of `asyncClear` or
+`asyncUpdate`for A and B are not determined.
+
+Although the state access methods are executed out of order, this not mean that all the user code
+are run in parallel. The user code in the `processElement`, `flatMap` or `thenXXxx` methods
+following the state access methods will be executed in a single thread (the task thread). So there
+is no concurrency issue for the user code.
+
+Typically, you don't need to worry about the execution order of the state access methods, but there
+is still some rules the Flink will ensure:
+* The execution order of user code entry `flatMap` for same-key elements are invoked strictly in
+order of element arrival.
+* The consumers or functions passed to the `thenXXxx` methods are executed in the order they are
+chained. If they are not chained, or there are multiple chains, the order is not guaranteed.
+
+### Best practice of asynchronous APIs
+
+The asynchronous APIs are designed to be more efficient and more powerful than the synchronous APIs.
+There are some best practices that you should follow when using the asynchronous APIs:
+* **Avoid mixing synchronous and asynchronous state access**
+* Use chaining of `thenXXxx` methods to handle the result of the state access and then gives another
+ state access or result. Divide the logic into multiple steps split by `thenXXxx` methods.
+* Avoid accessing mutable members of the user function (`RichFlatMapFunction`). Since the state access
+ methods are executed out of order, the mutable members may be accessed in unpredictable order.
+ Instead, use the result of the state access to pass the data between different steps. The
+ `StateFutureUtils.completedFuture` or `thenApply` method can be used to pass the data. Or use
+ a captured container (`AtomicReference`) which is initialized for each invoke of the `flatMap`
+ to share between lambdas.
+
+
### State Time-To-Live (TTL)
A *time-to-live* (TTL) can be assigned to the keyed state of any type. If a TTL is configured and a
@@ -336,7 +329,7 @@ expire independently.
In order to use state TTL one must first build a `StateTtlConfig` configuration object. The TTL
functionality can then be enabled in any state descriptor by passing the configuration:
-{{< tabs "43221f5b-b56c-42e5-833e-ed416e9d9b4e" >}}
+{{< tabs "43221f0b-b56c-42e5-833e-ed416e5f9b4e" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -353,38 +346,6 @@ ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text
stateDescriptor.enableTimeToLive(ttlConfig);
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-import org.apache.flink.api.common.state.ValueStateDescriptor
-import java.time.Duration
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
- .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
- .build
-
-val stateDescriptor = new ValueStateDescriptor[String]("text state", classOf[String])
-stateDescriptor.enableTimeToLive(ttlConfig)
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.common.typeinfo import Types
-from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite) \
- .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired) \
- .build()
-
-state_descriptor = ValueStateDescriptor("text state", Types.STRING())
-state_descriptor.enable_time_to_live(ttl_config)
-```
-{{< /tab >}}
{{< /tabs >}}
The configuration has several options to consider:
@@ -420,14 +381,15 @@ Another option `ReturnExpiredIfNotCleanedUp` allows to return the expired state
- The state backends store the timestamp of the last modification along with the user value,
which means that enabling this feature increases consumption of state storage.
Heap state backend stores an additional Java object with a reference to the user state object
-and a primitive long value in memory. The RocksDB state backend adds 8 bytes per stored value, list entry or map entry.
+and a primitive long value in memory. The RocksDB/ForSt state backend adds 8 bytes per stored value,
+list entry or map entry.
- Only TTLs in reference to *processing time* are currently supported.
- Trying to restore state, which was previously configured without TTL, using TTL enabled descriptor or vice versa
will lead to compatibility failure and `StateMigrationException`.
-- The TTL configuration is not part of check- or savepoints but rather a way of how Flink treats it in the currently running job.
+- The TTL configuration is not part of checkpoints or savepoints but rather a way of how Flink treats it in the currently running job.
- It is not recommended to restore checkpoint state with adjusting the ttl from a short value to a long value,
which may cause potential data errors.
@@ -442,7 +404,7 @@ If the serializer does not support null values, it can be wrapped with `Nullable
By default, expired values are explicitly removed on read, such as `ValueState#value`, and periodically garbage collected
in the background if supported by the configured state backend. Background cleanup can be disabled in the `StateTtlConfig`:
-{{< tabs "dfad4bf4-84ec-49a9-8372-c63b59122603" >}}
+{{< tabs "df774bf4-84ec-49a9-8372-c63b59122608" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -452,149 +414,25 @@ StateTtlConfig ttlConfig = StateTtlConfig
.build();
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .disableCleanupInBackground
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .disable_cleanup_in_background() \
- .build()
-```
-{{< /tab >}}
{{< /tabs >}}
-For more fine-grained control over some special cleanup in background, you can configure it separately as described below.
-Currently, heap state backend relies on incremental cleanup and RocksDB backend uses compaction filter for background cleanup.
-
-##### Cleanup in full snapshot
-
-Additionally, you can activate the cleanup at the moment of taking the full state snapshot which
-will reduce its size. The local state is not cleaned up under the current implementation
-but it will not include the removed expired state in case of restoration from the previous snapshot.
-It can be configured in `StateTtlConfig`:
-
-{{< tabs "ad8a4c67-222c-4beb-9547-ca5f8085695c" >}}
-{{< tab "Java" >}}
-```java
-import org.apache.flink.api.common.state.StateTtlConfig;
-import java.time.Duration;
-
-StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupFullSnapshot()
- .build();
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-import java.time.Duration
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupFullSnapshot
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_full_snapshot() \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-This option is not applicable for the incremental checkpointing in the RocksDB state backend.
-
-{{< hint info >}}
-For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
-{{< /hint >}}
-
-##### Incremental cleanup
-
-Another option is to trigger cleanup of some state entries incrementally.
-The trigger can be a callback from each state access or/and each record processing.
-If this cleanup strategy is active for certain state,
-The storage backend keeps a lazy global iterator for this state over all its entries.
-Every time incremental cleanup is triggered, the iterator is advanced.
-The traversed state entries are checked and expired ones are cleaned up.
+{{ <hint info> }}
+Currently the ForSt State Backend only cleanup the state in compaction process. If you are using
+a state backend other than ForSt and want to read about other cleanup strategies, please refer to the
+[State V1 documentation]({{< ref "docs/dev/datastream/fault-tolerance/state#cleanup-of-expired-state" >}})
+about the expired state cleanup.
+{{ </hint > }
-This feature can be configured in `StateTtlConfig`:
+##### Cleanup during compaction
-{{< tabs "e8c96bb9-6485-4f86-8fdb-5e538e538827" >}}
-{{< tab "Java" >}}
-```java
-import org.apache.flink.api.common.state.StateTtlConfig;
- StateTtlConfig ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupIncrementally(10, true)
- .build();
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupIncrementally(10, true)
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_incrementally(10, True) \
- .build()
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-This strategy has two parameters. The first one is number of checked state entries per each cleanup triggering.
-It is always triggered per each state access.
-The second parameter defines whether to trigger cleanup additionally per each record processing.
-The default background cleanup for heap backend checks 5 entries without cleanup per record processing.
-
-**Notes:**
-- If no access happens to the state or no records are processed, expired state will persist.
-- Time spent for the incremental cleanup increases record processing latency.
-- At the moment incremental cleanup is implemented only for Heap state backend. Setting it for RocksDB will have no effect.
-- If heap state backend is used with synchronous snapshotting, the global iterator keeps a copy of all keys
-while iterating because of its specific implementation which does not support concurrent modifications.
-Enabling of this feature will increase memory consumption then. Asynchronous snapshotting does not have this problem.
-- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
-
-##### Cleanup during RocksDB compaction
-
-If the RocksDB state backend is used, a Flink specific compaction filter will be called for the background cleanup.
-RocksDB periodically runs asynchronous compactions to merge state updates and reduce storage.
+If the ForSt state backend is used, a Flink specific compaction filter will be called for the background cleanup.
+ForSt periodically runs asynchronous compactions to merge state updates and reduce storage.
Flink compaction filter checks expiration timestamp of state entries with TTL
-and excludes expired values.
+and excludes expired values.
This feature can be configured in `StateTtlConfig`:
-{{< tabs "4db505be-2c9c-4ef7-aa01-74763968b7a1" >}}
+{{< tabs "4db505be-2c9c-6ef7-aa01-74763968b7a9" >}}
{{< tab "Java" >}}
```java
import org.apache.flink.api.common.state.StateTtlConfig;
@@ -605,40 +443,18 @@ StateTtlConfig ttlConfig = StateTtlConfig
.build();
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-import org.apache.flink.api.common.state.StateTtlConfig
-
-val ttlConfig = StateTtlConfig
- .newBuilder(Duration.ofSeconds(1))
- .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1))
- .build
-```
-{{< /tab >}}
-{{< tab "Python" >}}
-```python
-from pyflink.common import Duration
-from pyflink.common.time import Time
-from pyflink.datastream.state import StateTtlConfig
-
-ttl_config = StateTtlConfig \
- .new_builder(Duration.ofSeconds(1)) \
- .cleanup_in_rocksdb_compact_filter(1000, Duration.of_hours(1)) \
- .build()
-```
-{{< /tab >}}
{{< /tabs >}}
-RocksDB compaction filter will query current timestamp, used to check expiration, from Flink every time
+ForSt compaction filter will query current timestamp, used to check expiration, from Flink every time
after processing certain number of state entries.
-You can change it and pass a custom value to
-`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
-Updating the timestamp more often can improve cleanup speed
+You can change it and pass a custom value to
+`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries)` method.
+Updating the timestamp more often can improve cleanup speed
but it decreases compaction performance because it uses JNI call from native code.
-The default background cleanup for RocksDB backend queries the current timestamp each time 1000 entries have been processed.
+The default background cleanup for ForSt backend queries the current timestamp each time 1000 entries have been processed.
-Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed.
-Files older than this value will be picked up for compaction, and re-written to the same level as they were before.
+Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed.
+Files older than this value will be picked up for compaction, and re-written to the same level as they were before.
It makes sure a file goes through compaction filters periodically.
You can change it and pass a custom value to
`StateTtlConfig.newBuilder(...).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Duration periodicCompactionTime)` method.
@@ -646,42 +462,23 @@ The default value of Periodic compaction seconds is 30 days.
You could set it to 0 to turn off periodic compaction or set a small value to speed up expired state entries cleanup, but it
would trigger more compactions.
-You can activate debug logs from the native code of RocksDB filter
+You can activate debug logs from the native code of ForSt filter
by activating debug level for `FlinkCompactionFilter`:
-`log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG`
+`log4j.logger.org.forstdb.FlinkCompactionFilter=DEBUG`
-**Notes:**
-- Calling of TTL filter during compaction slows it down.
-The TTL filter has to parse timestamp of last access and check its expiration
-for every stored state entry per key which is being compacted.
-In case of collection state type (list or map) the check is also invoked per stored element.
+**Notes:**
+- Calling of TTL filter during compaction slows it down.
+ The TTL filter has to parse timestamp of last access and check its expiration
+ for every stored state entry per key which is being compacted.
+ In case of collection state type (list or map) the check is also invoked per stored element.
- If this feature is used with a list state which has elements with non-fixed byte length,
-the native TTL filter has to call additionally a Flink java type serializer of the element over JNI per each state entry
-where at least the first element has expired to determine the offset of the next unexpired element.
-- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
-e.g. after restart from savepoint.
+ the native TTL filter has to call additionally a Flink java type serializer of the element over JNI per each state entry
+ where at least the first element has expired to determine the offset of the next unexpired element.
+- For existing jobs, this cleanup strategy can be activated or deactivated anytime in `StateTtlConfig`,
+ e.g. after restart from savepoint.
- Periodic compaction could only work when TTL is enabled.
-### State in the Scala DataStream API
-
-In addition to the interface described above, the Scala API has shortcuts for stateful
-`map()` or `flatMap()` functions with a single `ValueState` on `KeyedStream`. The user function
-gets the current value of the `ValueState` in an `Option` and must return an updated value that
-will be used to update the state.
-
-```scala
-val stream: DataStream[(String, Int)] = ...
-
-val counts: DataStream[(String, Int)] = stream
- .keyBy(_._1)
- .mapWithState((in: (String, Int), count: Option[Int]) =>
- count match {
- case Some(c) => ( (in._1, c), Some(c + in._2) )
- case None => ( (in._1, 0), Some(in._2) )
- })
-```
-
## Operator State
*Operator State* (or *non-keyed state*) is state that is bound to one
@@ -697,8 +494,6 @@ In a typical stateful Flink Application you don't need operators state. It is
mostly a special type of state that is used in source/sink implementations and
scenarios where you don't have a key by which state can be partitioned.
-**Notes:** Operator state is still not supported in Python DataStream API.
-
## Broadcast State
*Broadcast State* is a special type of *Operator State*. It was introduced to
@@ -762,7 +557,7 @@ Below is an example of a stateful `SinkFunction` that uses `CheckpointedFunction
to buffer elements before sending them to the outside world. It demonstrates
the basic even-split redistribution list state:
-{{< tabs "5b99313c-1801-486c-a085-ce66882a2e49" >}}
+{{< tabs "5b99663c-1801-486c-a085-ce66882a2k49" >}}
{{< tab "Java" >}}
```java
public class BufferingSink
@@ -814,49 +609,6 @@ public class BufferingSink
}
```
{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class BufferingSink(threshold: Int = 0)
- extends SinkFunction[(String, Int)]
- with CheckpointedFunction {
-
- @transient
- private var checkpointedState: ListState[(String, Int)] = _
-
- private val bufferedElements = ListBuffer[(String, Int)]()
-
- override def invoke(value: (String, Int), context: Context): Unit = {
- bufferedElements += value
- if (bufferedElements.size >= threshold) {
- for (element <- bufferedElements) {
- // send it to the sink
- }
- bufferedElements.clear()
- }
- }
-
- override def snapshotState(context: FunctionSnapshotContext): Unit = {
- checkpointedState.update(bufferedElements.asJava)
- }
-
- override def initializeState(context: FunctionInitializationContext): Unit = {
- val descriptor = new ListStateDescriptor[(String, Int)](
- "buffered-elements",
- TypeInformation.of(new TypeHint[(String, Int)]() {})
- )
-
- checkpointedState = context.getOperatorStateStore.getListState(descriptor)
-
- if(context.isRestored) {
- for(element <- checkpointedState.get().asScala) {
- bufferedElements += element
- }
- }
- }
-
-}
-```
-{{< /tab >}}
{{< /tabs >}}
The `initializeState` method takes as argument a `FunctionInitializationContext`. This is used to initialize
@@ -879,18 +631,6 @@ ListStateDescriptor<Tuple2<String, Integer>> descriptor =
checkpointedState = context.getOperatorStateStore().getListState(descriptor);
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-
-val descriptor = new ListStateDescriptor[(String, Long)](
- "buffered-elements",
- TypeInformation.of(new TypeHint[(String, Long)]() {})
-)
-
-checkpointedState = context.getOperatorStateStore.getListState(descriptor)
-
-```
{{< /tab >}}
{{< /tabs >}}
The naming convention of the state access methods contain its redistribution
@@ -909,110 +649,16 @@ of all objects included by the previous checkpoint, and is then filled with the
As a side note, the keyed state can also be initialized in the `initializeState()` method. This can be done
using the provided `FunctionInitializationContext`.
-### Stateful Source Functions
-
-Stateful sources require a bit more care as opposed to other operators.
-In order to make the updates to the state and output collection atomic (required for exactly-once semantics
-on failure/recovery), the user is required to get a lock from the source's context.
-
-{{< tabs "aa821e89-7cc6-4c1e-be5b-90bfa0df6036" >}}
-{{< tab "Java" >}}
-```java
-public static class CounterSource
- extends RichParallelSourceFunction<Long>
- implements CheckpointedFunction {
-
- /** current offset for exactly once semantics */
- private Long offset = 0L;
-
- /** flag for job cancellation */
- private volatile boolean isRunning = true;
-
- /** Our state object. */
- private ListState<Long> state;
-
- @Override
- public void run(SourceContext<Long> ctx) {
- final Object lock = ctx.getCheckpointLock();
-
- while (isRunning) {
- // output and state update are atomic
- synchronized (lock) {
- ctx.collect(offset);
- offset += 1;
- }
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
- state = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
- "state",
- LongSerializer.INSTANCE));
-
- // restore any state that we might already have to our fields, initialize state
- // is also called in case of restore.
- for (Long l : state.get()) {
- offset = l;
- }
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- state.update(Collections.singletonList(offset));
- }
-}
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-```scala
-class CounterSource
- extends RichParallelSourceFunction[Long]
- with CheckpointedFunction {
-
- @volatile
- private var isRunning = true
-
- private var offset = 0L
- private var state: ListState[Long] = _
-
- override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
- val lock = ctx.getCheckpointLock
-
- while (isRunning) {
- // output and state update are atomic
- lock.synchronized({
- ctx.collect(offset)
-
- offset += 1
- })
- }
- }
+{{< top >}}
- override def cancel(): Unit = isRunning = false
-
- override def initializeState(context: FunctionInitializationContext): Unit = {
- state = context.getOperatorStateStore.getListState(
- new ListStateDescriptor[Long]("state", classOf[Long]))
+## Migrate from the Old State API
- for (l <- state.get().asScala) {
- offset = l
- }
- }
+It is very easy to migrate from the old state API to the new one. Please take the following steps:
- override def snapshotState(context: FunctionSnapshotContext): Unit = {
- state.update(java.util.Collections.singletonList(offset))
- }
-}
-```
-{{< /tab >}}
-{{< /tabs >}}
-
-Some operators might need the information when a checkpoint is fully acknowledged by Flink to communicate that with the outside world. In this case see the `org.apache.flink.api.common.state.CheckpointListener` interface.
-
-{{< top >}}
+1. Invoke `enableAsyncState()` on the `KeyedStream` to enable the new state API.
+2. Replace the `StateDescriptor`s with the new ones under `v2` package. Also, replace the old state
+handles with the new ones under `v2` package.
+3. Rewrite the old state access methods with the new asynchronous ones.
+4. It is recommended to use ForSt State Backend for the new state API, which could perform
+async state access. Other state backends only support synchronous execution of state access,
+although it can be used with the new state API.
(flink) 02/05: [FLINK-36439][docs] Documents for ForSt State Backend
Posted by za...@apache.org.
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 381be9cf9ec628f255e13962d2601c858303f10f
Author: Zakelly <za...@gmail.com>
AuthorDate: Thu Feb 6 12:49:23 2025 +0800
[FLINK-36439][docs] Documents for ForSt State Backend
---
docs/content.zh/docs/deployment/config.md | 30 ++-
docs/content.zh/docs/ops/metrics.md | 7 +-
docs/content/docs/deployment/config.md | 30 ++-
docs/content/docs/ops/metrics.md | 7 +-
docs/content/docs/ops/state/large_state_tuning.md | 16 +-
docs/content/docs/ops/state/state_backends.md | 57 ++++-
.../shortcodes/generated/expert_forst_section.html | 54 +++++
.../forst_configurable_configuration.html | 168 ++++++++++++++
.../shortcodes/generated/forst_configuration.html | 120 ++++++++++
.../forst_native_metric_configuration.html | 246 +++++++++++++++++++++
.../generated/state_backend_forst_section.html | 66 ++++++
.../flink/annotation/docs/Documentation.java | 2 +
flink-docs/pom.xml | 5 +
.../configuration/ConfigOptionsDocGenerator.java | 2 +
.../docs/util/ConfigurationOptionLocator.java | 3 +
.../state/forst/ForStConfigurableOptions.java | 52 +++--
.../state/forst/ForStNativeMetricOptions.java | 84 +++----
.../org/apache/flink/state/forst/ForStOptions.java | 64 ++++--
18 files changed, 917 insertions(+), 96 deletions(-)
diff --git a/docs/content.zh/docs/deployment/config.md b/docs/content.zh/docs/deployment/config.md
index 8fbf550db8f..8922586e2b3 100644
--- a/docs/content.zh/docs/deployment/config.md
+++ b/docs/content.zh/docs/deployment/config.md
@@ -160,7 +160,7 @@ These values are configured as memory sizes, for example *1536m* or *2g*.
You can configure checkpointing directly in code within your Flink job or application. Putting these values here in the configuration defines them as defaults in case the application does not configure anything.
- - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap` or `rocksdb`.
+ - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap`, `rocksdb` or `forst`.
- `execution.checkpointing.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*.
- `execution.checkpointing.savepoint-dir`: The default directory for savepoints. Takes a path URI, similar to `execution.checkpointing.dir`.
- `execution.checkpointing.interval`: The base interval setting. To enable checkpointing, you need to set this value larger than 0.
@@ -352,6 +352,12 @@ These are the options commonly needed to configure the RocksDB state backend. Se
{{< generated/state_backend_rocksdb_section >}}
+### ForSt State Backend
+
+These are the options commonly needed to configure the ForSt state backend. See the [Advanced ForSt Backend Section](#advanced-forst-state-backends-options) for options necessary for advanced low level configurations and trouble-shooting.
+
+{{< generated/state_backend_forst_section >}}
+
----
----
@@ -374,6 +380,16 @@ Enabling RocksDB's native metrics may cause degraded performance and should be s
{{< generated/rocksdb_native_metric_configuration >}}
+### ForSt Native Metrics
+
+ForSt has similar native metric mechanism to RocksDB.
+
+{{< hint warning >}}
+Enabling ForSt's native metrics may cause degraded performance and should be set carefully.
+{{< /hint >}}
+
+{{< generated/forst_native_metric_configuration >}}
+
----
----
@@ -474,6 +490,12 @@ Advanced options to tune RocksDB and RocksDB checkpoints.
{{< generated/expert_rocksdb_section >}}
+### Advanced ForSt State Backends Options
+
+Advanced options to tune ForSt and ForSt checkpoints.
+
+{{< generated/expert_forst_section >}}
+
### State Changelog Options
Please refer to [State Backends]({{< ref "docs/ops/state/state_backends#enabling-changelog" >}}) for information on
@@ -484,13 +506,17 @@ using State Changelog. {{< generated/state_changelog_section >}}
These settings take effect when the `state.changelog.storage` is set to `filesystem` (see [above](#state-changelog-storage)).
{{< generated/fs_state_changelog_configuration >}}
-**RocksDB Configurable Options**
+### RocksDB Configurable Options
These options give fine-grained control over the behavior and resources of ColumnFamilies.
With the introduction of `state.backend.rocksdb.memory.managed` and `state.backend.rocksdb.memory.fixed-per-slot` (Apache Flink 1.10), it should be only necessary to use the options here for advanced performance tuning. These options here can also be specified in the application program via `RocksDBStateBackend.setRocksDBOptions(RocksDBOptionsFactory)`.
{{< generated/rocksdb_configurable_configuration >}}
+### ForSt State Backend Configurable Options
+
+{{< generated/forst_configurable_configuration >}}
+
### Advanced Fault Tolerance Options
*These parameters can help with problems related to failover and to components erroneously considering each other as failed.*
diff --git a/docs/content.zh/docs/ops/metrics.md b/docs/content.zh/docs/ops/metrics.md
index 852893b2c52..bb3fc9c9317 100644
--- a/docs/content.zh/docs/ops/metrics.md
+++ b/docs/content.zh/docs/ops/metrics.md
@@ -1541,7 +1541,12 @@ Note that for failed checkpoints, metrics are updated on a best efforts basis an
### RocksDB
Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#rocksdb-native-metrics)
-### ForStDB
+### ForSt
+
+Certain ForSt native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#forst-native-metrics)
+
+Besides that, we support the following metrics:
+
<table class="table table-bordered">
<thead>
<tr>
diff --git a/docs/content/docs/deployment/config.md b/docs/content/docs/deployment/config.md
index 7c149528963..a804fdc34a3 100644
--- a/docs/content/docs/deployment/config.md
+++ b/docs/content/docs/deployment/config.md
@@ -160,7 +160,7 @@ These values are configured as memory sizes, for example *1536m* or *2g*.
You can configure checkpointing directly in code within your Flink job or application. Putting these values here in the configuration defines them as defaults in case the application does not configure anything.
- - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap` or `rocksdb`.
+ - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap`, `rocksdb` or `forst`.
- `execution.checkpointing.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*.
- `execution.checkpointing.savepoint-dir`: The default directory for savepoints. Takes a path URI, similar to `execution.checkpointing.dir`.
- `execution.checkpointing.interval`: The base interval setting. To enable checkpointing, you need to set this value larger than 0.
@@ -354,6 +354,12 @@ These are the options commonly needed to configure the RocksDB state backend. Se
{{< generated/state_backend_rocksdb_section >}}
+### ForSt State Backend
+
+These are the options commonly needed to configure the ForSt state backend. See the [Advanced ForSt Backend Section](#advanced-forst-state-backends-options) for options necessary for advanced low level configurations and trouble-shooting.
+
+{{< generated/state_backend_forst_section >}}
+
----
----
@@ -376,6 +382,16 @@ Enabling RocksDB's native metrics may cause degraded performance and should be s
{{< generated/rocksdb_native_metric_configuration >}}
+### ForSt Native Metrics
+
+ForSt has similar native metric mechanism to RocksDB.
+
+{{< hint warning >}}
+Enabling ForSt's native metrics may cause degraded performance and should be set carefully.
+{{< /hint >}}
+
+{{< generated/forst_native_metric_configuration >}}
+
----
----
@@ -476,6 +492,12 @@ Advanced options to tune RocksDB and RocksDB checkpoints.
{{< generated/expert_rocksdb_section >}}
+### Advanced ForSt State Backends Options
+
+Advanced options to tune ForSt and ForSt checkpoints.
+
+{{< generated/expert_forst_section >}}
+
### State Changelog Options
Please refer to [State Backends]({{< ref "docs/ops/state/state_backends#enabling-changelog" >}}) for information on
@@ -486,13 +508,17 @@ using State Changelog. {{< generated/state_changelog_section >}}
These settings take effect when the `state.changelog.storage` is set to `filesystem` (see [above](#state-changelog-storage)).
{{< generated/fs_state_changelog_configuration >}}
-**RocksDB Configurable Options**
+### RocksDB Configurable Options
These options give fine-grained control over the behavior and resources of ColumnFamilies.
With the introduction of `state.backend.rocksdb.memory.managed` and `state.backend.rocksdb.memory.fixed-per-slot` (Apache Flink 1.10), it should be only necessary to use the options here for advanced performance tuning. These options here can also be specified in the application program via `RocksDBStateBackend.setRocksDBOptions(RocksDBOptionsFactory)`.
{{< generated/rocksdb_configurable_configuration >}}
+### ForSt State Backend Configurable Options
+
+{{< generated/forst_configurable_configuration >}}
+
### Advanced Fault Tolerance Options
*These parameters can help with problems related to failover and to components erroneously considering each other as failed.*
diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md
index f091e2d9d31..70a99457e80 100644
--- a/docs/content/docs/ops/metrics.md
+++ b/docs/content/docs/ops/metrics.md
@@ -1531,7 +1531,12 @@ Note that for failed checkpoints, metrics are updated on a best efforts basis an
### RocksDB
Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#rocksdb-native-metrics)
-### ForStDB
+### ForSt
+
+Certain ForSt native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#forst-native-metrics)
+
+Besides that, we support the following metrics:
+
<table class="table table-bordered">
<thead>
<tr>
diff --git a/docs/content/docs/ops/state/large_state_tuning.md b/docs/content/docs/ops/state/large_state_tuning.md
index e783d2b56c1..d743a0f05cf 100644
--- a/docs/content/docs/ops/state/large_state_tuning.md
+++ b/docs/content/docs/ops/state/large_state_tuning.md
@@ -97,13 +97,27 @@ the same time. For applications with large state in Flink, this often ties up to
When a savepoint is manually triggered, it may be in process concurrently with an ongoing checkpoint.
-## Tuning RocksDB
+## Tuning RocksDB or ForSt
The state storage workhorse of many large scale Flink streaming applications is the *RocksDB State Backend*.
The backend scales well beyond main memory and reliably stores large [keyed state]({{< ref "docs/dev/datastream/fault-tolerance/state" >}}).
+If you are handling very large state, even exceeding the local disk space of the TaskManagers, you may want to consider
+using the disaggregated state store [ForStStateBackend]({{< ref "docs/deployment/config" >}}#forst-state-backend).
+This backend stores the state in a separate storage system, such as HDFS or S3, and only keeps the
+state metadata and cache in the TaskManagers. And the [State API V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}})
+is also recommended to cooperate with ForStStateBackend for large state applications.
+
RocksDB's performance can vary with configuration, this section outlines some best-practices for tuning jobs that use the RocksDB State Backend.
+{{< hint info >}}
+The design of ForSt is very similar to RocksDB, and the configurable options are almost the same,
+so you can refer to following sections to configure ForSt.
+
+The following article is introduced from the perspective of RocksDB. If you want to configure ForSt
+in a similar way, you need to use the corresponding configuration under ForSt.
+{{< /hint >}}
+
### Incremental Checkpoints
When it comes to reducing the time that checkpoints take, activating incremental checkpoints should be one of the first considerations.
diff --git a/docs/content/docs/ops/state/state_backends.md b/docs/content/docs/ops/state/state_backends.md
index 6594018422f..b3cb7462333 100644
--- a/docs/content/docs/ops/state/state_backends.md
+++ b/docs/content/docs/ops/state/state_backends.md
@@ -95,12 +95,50 @@ Certain RocksDB native metrics are available but disabled by default, you can fi
The total memory amount of RocksDB instance(s) per slot can also be bounded, please refer to documentation [here]({{< ref "docs/ops/state/large_state_tuning" >}}#bounding-rocksdb-memory-usage) for details.
+## The ForStStateBackend
+
+The *ForStStateBackend* is a state backend that is based on [ForSt project](https://github.com/ververica/ForSt),
+which is also a LSM-tree structured key-value store and built on top of the RocksDB.
+It is designed to provide a more efficient way to store and access state in Flink applications.
+Most importantly, it can hold its sst files on remote file systems that Flink supports, such as HDFS, S3, etc.
+This allows Flink to scale the state size beyond the local disk capacity of the TaskManager.
+Moreover, by putting the sst files on remote file systems, it can also provide a more lightweight
+way to perform checkpoint and recovery.
+
+The ForStStateBackend is still in the experimental stage and is not fully available for production.
+It always performs asynchronous incremental snapshots.
+
+The ForStStateBackend is encouraged for:
+
+- Jobs with very large state, long windows, large key/value states. Local disk may not be enough to
+store the state.
+- All high-availability setups.
+- Asynchronous state access is preferred. Since the ForStStateBackend is the only one supporting
+asynchronous state access.
+- Jobs that require lightweight checkpoint and recovery, such as cloud-native applications.
+
+Limitations of the ForStStateBackend (for now):
+
+- Same as EmbeddedRocksDBStateBackend, the maximum supported size per key and per value is 2^31 bytes each.
+- Does not support canonical savepoint, full snapshot, changelog and file-merging checkpoints.
+Always perform incremental snapshots.
+
+Compared with EmbeddedRocksDBStateBackend, ForStStateBackend stores data on remote file system, thus
+the amount of state that you can keep is unlimited. The local disk of TaskManager is only used to
+store cache of file, to provide better performance. Note that when most of the active state is on
+remote file system, the performance of state access may be affected by the network latency. Flink
+introduces asynchronous state access to mitigate this issue. If you are using the asynchronous state
+methods in State API V2, you can benefit from the asynchronous state access. To get familiar with the
+State API V2, please refer to the [State API V2 documentation]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+
## Choose The Right State Backend
When deciding between `HashMapStateBackend` and `RocksDB`, it is a choice between performance and scalability.
`HashMapStateBackend` is very fast as each state access and update operates on objects on the Java heap; however, state size is limited by available memory within the cluster.
-On the other hand, `RocksDB` can scale based on available disk space and is the only state backend to support incremental snapshots.
+On the other hand, `RocksDB` can scale based on available disk space.
However, each state access and update requires (de-)serialization and potentially reading from disk which leads to average performance that is an order of magnitude slower than the memory state backends.
+If you are handling very large state even exceeding the available disk space,
+or you prefer a fast rescale under cloud-native setup, you should consider using `ForStStateBackend`.
{{< hint info >}}
In Flink 1.13 we unified the binary format of Flink's savepoints. That means you can take a savepoint and then restore from it using a different state backend.
@@ -150,8 +188,18 @@ If you want to use the `EmbeddedRocksDBStateBackend` in your IDE or configure it
</dependency>
```
+Same for `ForStStateBackend`:
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-forst</artifactId>
+ <version>{{< version >}}</version>
+ <scope>provided</scope>
+</dependency>
+```
+
{{< hint info >}}
-Since RocksDB is part of the default Flink distribution, you do not need this dependency if you are not using any RocksDB code in your job and configure the state backend via `state.backend.type` and further [checkpointing]({{< ref "docs/deployment/config" >}}#checkpointing) and [RocksDB-specific]({{< ref "docs/deployment/config" >}}#rocksdb-state-backend) parameters in your [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}).
+Since RocksDB and ForSt is part of the default Flink distribution, you do not need this dependency if you are not using any RocksDB code in your job and configure the state backend via `state.backend.type` and further [checkpointing]({{< ref "docs/deployment/config" >}}#checkpointing) and [RocksDB-specific]({{< ref "docs/deployment/config" >}}#rocksdb-state-backend) or [ForSt-specific]({{< ref "docs/deployment/config" >}}#forst-state-backend) parameters in your [Flink configuration file] [...]
{{< /hint >}}
@@ -159,9 +207,10 @@ Since RocksDB is part of the default Flink distribution, you do not need this de
A default state backend can be configured in the [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}), using the configuration key `state.backend.type`.
-Possible values for the config entry are *hashmap* (HashMapStateBackend), *rocksdb* (EmbeddedRocksDBStateBackend), or the fully qualified class
+Possible values for the config entry are *hashmap* (HashMapStateBackend), *rocksdb* (EmbeddedRocksDBStateBackend), *forst* (ForStStateBackend) or the fully qualified class
name of the class that implements the state backend factory {{< gh_link file="flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendFactory.java" name="StateBackendFactory" >}},
-such as `org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackendFactory` for EmbeddedRocksDBStateBackend.
+such as `org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackendFactory` for EmbeddedRocksDBStateBackend
+and `org.apache.flink.state.forst.ForStStateBackendFactory` for ForStStateBackend.
The `execution.checkpointing.dir` option defines the directory to which all backends write checkpoint data and meta data files.
You can find more details about the checkpoint directory structure [here]({{< ref "docs/ops/state/checkpoints" >}}#directory-structure).
diff --git a/docs/layouts/shortcodes/generated/expert_forst_section.html b/docs/layouts/shortcodes/generated/expert_forst_section.html
new file mode 100644
index 00000000000..ae289ad4582
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/expert_forst_section.html
@@ -0,0 +1,54 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to let the task thread be the coordinator thread responsible for distributing requests. If set to 'true', the task thread will be responsible for distributing requests, otherwise, a dedicated coordinator thread will be used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-write</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to let write requests be executed within the coordinator thread. If set to 'true', write requests will be executed within the coordinator thread, otherwise, a dedicated write thread will be used. The default value is 'true'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.local-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-slot</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per slot, shared among all ForSt instances.This option overrides the 'state.backend.forst.memory.managed' option.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-tm</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. This option only takes effect if 'state.backend.forst.memory.managed' is set to false and 'state.backend.forst.memory.fixed-per-slot' is not configured. If so, then each ForSt column family state has its own memory caches (as controlled by the column family options). The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the s [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.managed</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If set true, the ForSt state backend will automatically configure itself to use the managed memory budget of the task slot, and divide the memory over write buffers, indexes, block caches, etc.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.options-factory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions after loading ones from 'ForStConfigurableOptions' and pre-defined options.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_configurable_configuration.html b/docs/layouts/shortcodes/generated/forst_configurable_configuration.html
new file mode 100644
index 00000000000..a6aac0a00f4
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_configurable_configuration.html
@@ -0,0 +1,168 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.block.blocksize</h5></td>
+ <td style="word-wrap: break-word;">4 kb</td>
+ <td>MemorySize</td>
+ <td>The approximate size (in bytes) of user data packed per block. The default blocksize is '4KB'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.block.cache-size</h5></td>
+ <td style="word-wrap: break-word;">8 mb</td>
+ <td>MemorySize</td>
+ <td>The amount of the cache for data blocks in ForSt. The default block-cache size is '8MB'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.block.metadata-blocksize</h5></td>
+ <td style="word-wrap: break-word;">4 kb</td>
+ <td>MemorySize</td>
+ <td>Approximate size of partitioned metadata packed per block. Currently applied to indexes block when partitioned index/filters option is enabled. The default blocksize is '4KB'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.bloom-filter.bits-per-key</h5></td>
+ <td style="word-wrap: break-word;">10.0</td>
+ <td>Double</td>
+ <td>Bits per key that the bloom filter will use, this only takes effect when the bloom filter is used. The default value is 10.0.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.bloom-filter.block-based-mode</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If set 'true', ForSt will use block-based filter instead of full filter, this only takes effect when bloom filter is used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.filter.periodic-compaction-time</h5></td>
+ <td style="word-wrap: break-word;">30 d</td>
+ <td>Duration</td>
+ <td>Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed. Files older than this value will be picked up for compaction, and re-written to the same level as they were before. It makes sure a file goes through compaction filters periodically. 0 means turning off periodic compaction.The default value is '30 d' (30 days).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.filter.query-time-after-num-entries</h5></td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Long</td>
+ <td>Number of state entries to process by compaction filter before updating current timestamp. Updating the timestamp more often can improve cleanup speed, but it decreases compaction performance because it uses JNI calls from native code.The default value is '1000L'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.max-size-level-base</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>The upper-bound of the total size of level base files in bytes. The default value is '256 mb'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.target-file-size-base</h5></td>
+ <td style="word-wrap: break-word;">64 mb</td>
+ <td>MemorySize</td>
+ <td>The target file size for compaction, which determines a level-1 file size. The default value is '64 mb'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.use-dynamic-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, ForSt will pick target size of each level dynamically. From an empty key-value store, ForSt would make last level the base level, which means merging L0 data into the last level, until it exceeds max_bytes_for_level_base. And then repeat this process for second last level and so on. The default value is 'false'. For more information, please refer to <a href="https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true"> [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.style</h5></td>
+ <td style="word-wrap: break-word;">LEVEL</td>
+ <td><p>Enum</p></td>
+ <td>The specified compaction style for DB. Candidate compaction style is LEVEL, FIFO, UNIVERSAL or NONE, and Flink chooses 'LEVEL' as default style.<br /><br />Possible values:<ul><li>"LEVEL"</li><li>"UNIVERSAL"</li><li>"FIFO"</li><li>"NONE"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compression.per.level</h5></td>
+ <td style="word-wrap: break-word;">SNAPPY_COMPRESSION</td>
+ <td><p>List<Enum></p></td>
+ <td>A semicolon-separated list of Compression Type. Different levels can have different compression policies. In many cases, lower levels use fast compression algorithms, while higher levels with more data use slower but more effective compression algorithms. The N th element in the List corresponds to the compression type of the level N-1. When <code class="highlighter-rouge">state.backend.forst.compaction.level.use-dynamic-size</code> is true, compression_per_level[0] still [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.files.open</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Integer</td>
+ <td>The maximum number of open files (per stateful operator) that can be used by the ForSt, '-1' means no limit. The default value is '-1'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The directory for ForSt's information logging files. If empty (Flink default setting), log files will be in the same directory as the Flink log. If non-empty, this directory will be used and the data directory's absolute path will be used as the prefix of the log file name. If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.file-num</h5></td>
+ <td style="word-wrap: break-word;">4</td>
+ <td>Integer</td>
+ <td>The maximum number of files ForSt should keep for information logging (Default setting: 4).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.level</h5></td>
+ <td style="word-wrap: break-word;">INFO_LEVEL</td>
+ <td><p>Enum</p></td>
+ <td>The specified information logging level for ForSt. If unset, Flink will use <code class="highlighter-rouge">INFO_LEVEL</code>.<br />Note: ForSt info logs will not be written to the TaskManager logs and there is no rolling strategy, unless you configure <code class="highlighter-rouge">state.backend.forst.log.dir</code>, <code class="highlighter-rouge">state.backend.forst.log.max-file-size</code>, and <code class="highlighter-rouge">state.backend.forst.log.file-num</code> a [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.max-file-size</h5></td>
+ <td style="word-wrap: break-word;">25 mb</td>
+ <td>MemorySize</td>
+ <td>The maximum size of ForSt's file used for information logging. If the log files becomes larger than this, a new file will be created. If 0, all logs will be written to one log file. The default maximum file size is '25MB'. </td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.rescaling.use-delete-files-in-range</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, during rescaling, the deleteFilesInRange API will be invoked to clean up the useless key-values so that primary storage space can be reclaimed more promptly.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.restore-overlap-fraction-threshold</h5></td>
+ <td style="word-wrap: break-word;">0.0</td>
+ <td>Double</td>
+ <td>The threshold of overlap fraction between the state handle's key-group range and target key-group range. When restore base DB, only the handle which overlap fraction greater than or equal to threshold has a chance to be an initial handle. The default value is 0.0, there is always a handle will be selected for initialization. </td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.thread.num</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>The maximum number of concurrent background flush and compaction jobs (per stateful operator). The default value is '2'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.use-bloom-filter</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether every newly created SST file will contain a Bloom filter. Default 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.use-ingest-db-restore-mode</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>A recovery mode that directly clips and ingests multiple DBs during state recovery if the keys in the SST files does not exceed the declared key-group range.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.write-batch-size</h5></td>
+ <td style="word-wrap: break-word;">2 mb</td>
+ <td>MemorySize</td>
+ <td>The max size of the consumed memory for ForSt batch write, will flush just based on item count if this config set to 0.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.count</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>The maximum number of write buffers that are built up in memory. The default value is '2'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.number-to-merge</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The minimum number of write buffers that will be merged together before writing to storage. The default value is '1'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.size</h5></td>
+ <td style="word-wrap: break-word;">64 mb</td>
+ <td>MemorySize</td>
+ <td>The amount of data built up in memory (backed by an unsorted log on disk) before converting to a sorted on-disk files. The default writebuffer size is '64MB'.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_configuration.html b/docs/layouts/shortcodes/generated/forst_configuration.html
new file mode 100644
index 00000000000..56d6855ad82
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -0,0 +1,120 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.cache.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.reserve-size</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.size-based-limit</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The size-based capacity limit of cache, a non-positive number indicates that there is no limit. The default value is '-1', which means that the cache size is not limited by size.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to let the task thread be the coordinator thread responsible for distributing requests. If set to 'true', the task thread will be responsible for distributing requests, otherwise, a dedicated coordinator thread will be used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-write</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to let write requests be executed within the coordinator thread. If set to 'true', write requests will be executed within the coordinator thread, otherwise, a dedicated write thread will be used. The default value is 'true'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.read-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The number of threads used for read IO operations in the executor.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.write-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for write IO operations in the executor. Only valid when 'state.backend.forst.executor.inline-write' is false.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.local-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-slot</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per slot, shared among all ForSt instances.This option overrides the 'state.backend.forst.memory.managed' option.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-tm</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. This option only takes effect if 'state.backend.forst.memory.managed' is set to false and 'state.backend.forst.memory.fixed-per-slot' is not configured. If so, then each ForSt column family state has its own memory caches (as controlled by the column family options). The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the s [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.high-prio-pool-ratio</h5></td>
+ <td style="word-wrap: break-word;">0.1</td>
+ <td>Double</td>
+ <td>The fraction of cache memory that is reserved for high-priority data like index, filter, and compression dictionary blocks. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.memory.fixed-per-slot' are configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.managed</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If set true, the ForSt state backend will automatically configure itself to use the managed memory budget of the task slot, and divide the memory over write buffers, indexes, block caches, etc.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.partitioned-index-filters</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. The partitioned index/filter then uses the top-level index to load on demand into the block cache the partitions that are required to perform the index/filter query. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.mem [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.write-buffer-ratio</h5></td>
+ <td style="word-wrap: break-word;">0.5</td>
+ <td>Double</td>
+ <td>The maximum amount of memory that write buffers may take, as a fraction of the total shared memory. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.memory.fixed-per-slot' are configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.options-factory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions after loading ones from 'ForStConfigurableOptions' and pre-defined options.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.remote-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.cache-size</h5></td>
+ <td style="word-wrap: break-word;">128</td>
+ <td>Integer</td>
+ <td>The cache size per keyGroup of ForSt timer service factory. This option only has an effect when 'state.backend.forst.timer-service.factory' is configured to 'ForStDB'. Increasing this value can improve the performance of ForSt timer service, but consumes more heap memory at the same time.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.factory</h5></td>
+ <td style="word-wrap: break-word;">ForStDB</td>
+ <td><p>Enum</p></td>
+ <td>This determines the factory for timer service state implementation.<br /><br />Possible values:<ul><li>"HEAP": Heap-based</li><li>"ForStDB": Implementation based on RocksDB</li></ul></td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html b/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html
new file mode 100644
index 00000000000..96c522050a8
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html
@@ -0,0 +1,246 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.metrics.actual-delayed-write-rate</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the current actual delayed write rate. 0 means no delay.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.background-errors</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of background errors in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-capacity</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor block cache capacity.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-hit</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of block cache hit in ForSt (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-miss</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of block cache misses in ForSt (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-pinned-usage</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the memory size for the entries being pinned in block cache.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-usage</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the memory size for the entries residing in block cache.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-full-positive</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads not avoided by bloom full filter.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-full-true-positive</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads not avoided by bloom full filter and the data actually exists in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-useful</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads avoided by bloom filter.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bytes-read</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bytes-written</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.column-family-as-variable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to expose the column family as a variable for ForSt property based metrics.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Track pending compactions in ForSt. Returns 1 if a compaction is pending, 0 otherwise.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-read-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the bytes read during compaction in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-write-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the bytes written during compaction in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.cur-size-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active memtable in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.cur-size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active and unflushed immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-live-data-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate of the amount of live data in bytes (usually smaller than sst files size due to space amplification).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-num-keys</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate the number of keys in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-pending-compaction-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimated total number of bytes compaction needed to rewrite to get all levels down to under target size. Not valid for other compactions than level-based.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-table-readers-mem</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate the memory used for reading SST tables, excluding memory used in block cache (e.g. filter and index blocks) in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.is-write-stopped</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Track whether write has been stopped in ForSt. The metric will return 1 if write has been stopped, 0 otherwise.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.iter-bytes-read</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.live-sst-files-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total size (bytes) of all SST files belonging to the latest version. WARNING: may slow down online queries if there are too many files.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.mem-table-flush-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of pending memtable flushes in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-deletes-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of delete entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-deletes-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of delete entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-entries-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-entries-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-files-at-level</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of files at each level.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-immutable-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of immutable memtables in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-live-versions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor number of live versions. Version is an internal data structure. See ForSt file version_set.h for details. More live versions often mean more SST files are held from being deleted, by iterators or unfinished compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-running-compactions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of currently running compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-running-flushes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of currently running flushes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-snapshots</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of unfinished snapshots of the ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active, unflushed immutable, and pinned immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.stall-micros</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the writer wait duration for compaction or flush to finish in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.total-sst-files-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total size (bytes) of all SST files of all versions. WARNING: may slow down online queries if there are too many files.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/state_backend_forst_section.html b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
new file mode 100644
index 00000000000..29459c4157d
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
@@ -0,0 +1,66 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.cache.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.reserve-size</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.size-based-limit</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The size-based capacity limit of cache, a non-positive number indicates that there is no limit. The default value is '-1', which means that the cache size is not limited by size.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.read-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The number of threads used for read IO operations in the executor.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.write-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for write IO operations in the executor. Only valid when 'state.backend.forst.executor.inline-write' is false.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.partitioned-index-filters</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. The partitioned index/filter then uses the top-level index to load on demand into the block cache the partitions that are required to perform the index/filter query. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.mem [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.remote-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.cache-size</h5></td>
+ <td style="word-wrap: break-word;">128</td>
+ <td>Integer</td>
+ <td>The cache size per keyGroup of ForSt timer service factory. This option only has an effect when 'state.backend.forst.timer-service.factory' is configured to 'ForStDB'. Increasing this value can improve the performance of ForSt timer service, but consumes more heap memory at the same time.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.factory</h5></td>
+ <td style="word-wrap: break-word;">ForStDB</td>
+ <td><p>Enum</p></td>
+ <td>This determines the factory for timer service state implementation.<br /><br />Possible values:<ul><li>"HEAP": Heap-based</li><li>"ForStDB": Implementation based on RocksDB</li></ul></td>
+ </tr>
+ </tbody>
+</table>
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
index 8b766624152..e469fd8c386 100644
--- a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
@@ -77,6 +77,7 @@ public final class Documentation {
public static final String SECURITY_AUTH_ZOOKEEPER = "security_auth_zk";
public static final String STATE_BACKEND_ROCKSDB = "state_backend_rocksdb";
+ public static final String STATE_BACKEND_FORST = "state_backend_forst";
public static final String STATE_LATENCY_TRACKING = "state_latency_tracking";
@@ -95,6 +96,7 @@ public final class Documentation {
"expert_high_availability_k8s";
public static final String EXPERT_SECURITY_SSL = "expert_security_ssl";
public static final String EXPERT_ROCKSDB = "expert_rocksdb";
+ public static final String EXPERT_FORST = "expert_forst";
public static final String EXPERT_CLUSTER = "expert_cluster";
public static final String EXPERT_JOB_MANAGER = "expert_jobmanager";
diff --git a/flink-docs/pom.xml b/flink-docs/pom.xml
index efa4fdee02e..160f35d65d3 100644
--- a/flink-docs/pom.xml
+++ b/flink-docs/pom.xml
@@ -134,6 +134,11 @@ under the License.
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-forst</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-python</artifactId>
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
index 0af7bbcf9f0..a392ae3a0e5 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
@@ -260,6 +260,8 @@ public class ConfigOptionsDocGenerator {
@VisibleForTesting
static String toSnakeCase(String name) {
+ // We prefer lower case 'forst' for 'ForSt' instead of 'for_st'
+ name = name.replaceAll("(ForSt)([A-Z]|$)", "forst$2");
return name.replaceAll("(.)([A-Z][a-z])", "$1_$2").toLowerCase();
}
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
index 00e4bb65fc7..d3deb0fd777 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
@@ -68,6 +68,9 @@ public class ConfigurationOptionLocator {
new OptionsClassLocation(
"flink-state-backends/flink-statebackend-rocksdb",
"org.apache.flink.state.rocksdb"),
+ new OptionsClassLocation(
+ "flink-state-backends/flink-statebackend-forst",
+ "org.apache.flink.state.forst"),
new OptionsClassLocation(
"flink-table/flink-table-api-java", "org.apache.flink.table.api.config"),
new OptionsClassLocation("flink-python", "org.apache.flink.python"),
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
index 3b96bf89164..2e6c9aa19b3 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
@@ -39,6 +39,7 @@ import java.util.Set;
import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.configuration.description.LinkElement.link;
import static org.apache.flink.configuration.description.TextElement.code;
+import static org.apache.flink.configuration.description.TextElement.text;
import static org.apache.flink.state.forst.ForStOptions.CACHE_DIRECTORY;
import static org.apache.flink.state.forst.ForStOptions.CACHE_RESERVED_SIZE;
import static org.apache.flink.state.forst.ForStOptions.CACHE_SIZE_BASE_LIMIT;
@@ -81,7 +82,7 @@ public class ForStConfigurableOptions implements Serializable {
.intType()
.defaultValue(-1)
.withDescription(
- "The maximum number of open files (per stateful operator) that can be used by the DB, '-1' means no limit. "
+ "The maximum number of open files (per stateful operator) that can be used by the ForSt, '-1' means no limit. "
+ "The default value is '-1'.");
public static final ConfigOption<MemorySize> LOG_MAX_FILE_SIZE =
@@ -109,7 +110,7 @@ public class ForStConfigurableOptions implements Serializable {
"The directory for ForSt's information logging files. "
+ "If empty (Flink default setting), log files will be in the same directory as the Flink log. "
+ "If non-empty, this directory will be used and the data directory's absolute path will be used as the prefix of the log file name. "
- + "If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder as before.");
+ + "If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder.");
public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
key("state.backend.forst.log.level")
@@ -160,7 +161,7 @@ public class ForStConfigurableOptions implements Serializable {
.withDescription(
Description.builder()
.text(
- "If true, ForSt will pick target size of each level dynamically. From an empty DB, ")
+ "If true, ForSt will pick target size of each level dynamically. From an empty key-value store, ")
.text(
"ForSt would make last level the base level, which means merging L0 data into the last level, ")
.text(
@@ -182,27 +183,27 @@ public class ForStConfigurableOptions implements Serializable {
Description.builder()
.text(
"A semicolon-separated list of Compression Type. Different levels can have different "
- + "compression policies. In many cases, lower levels use fast compression algorithms,"
- + " while higher levels with more data use slower but more effective compression algorithms. "
- + "The N th element in the List corresponds to the compression type of the level N-1"
+ + "compression policies. In many cases, lower levels use fast compression algorithms, "
+ + "while higher levels with more data use slower but more effective compression algorithms. "
+ + "The N th element in the List corresponds to the compression type of the level N-1. "
+ "When %s is true, compression_per_level[0] still determines L0, but other "
- + "elements are based on the base level and may not match the level seen in the info log",
+ + "elements are based on the base level and may not match the level seen in the info log.",
code(USE_DYNAMIC_LEVEL_SIZE.key()))
.linebreak()
.text(
- "Note: If the List size is smaller than the level number, the undefined lower level uses the last Compression Type in the List")
+ "Note: If the List size is smaller than the level number, the undefined lower level uses the last Compression Type in the List.")
.linebreak()
.text(
- "Some commonly used compression algorithms for candidates include %s ,%s and %s",
+ "Some commonly used compression algorithms for candidates include %s ,%s and %s.",
code(NO_COMPRESSION.name()),
code(SNAPPY_COMPRESSION.name()),
code(LZ4_COMPRESSION.name()))
.linebreak()
.text(
- "The default value is %s, which means that all data uses the Snappy compression algorithm.",
+ "The default value is %s, which means that all data uses the Snappy compression algorithm. ",
code(SNAPPY_COMPRESSION.name()))
.text(
- "Likewise, if set to %s , means that all data is not compressed, which will achieve faster speed but will bring some space amplification.",
+ "Likewise, if set to %s , means that all data is not compressed, which will achieve faster speed but will bring some space amplification. ",
code(NO_COMPRESSION.name()))
.text(
"In addition, if we need to consider both spatial amplification and performance, we can also set it to '%s;%s;%s', which means that L0 and L1 data will not be compressed, and other data will be compressed using LZ4.",
@@ -216,16 +217,24 @@ public class ForStConfigurableOptions implements Serializable {
.memoryType()
.defaultValue(MemorySize.parse("64mb"))
.withDescription(
- "The target file size for compaction, which determines a level-1 file size. "
- + "The default value is '64MB'.");
+ Description.builder()
+ .text(
+ "The target file size for compaction, which determines a level-1 file size. "
+ + "The default value is '%s'.",
+ text(MemorySize.parse("64mb").toString()))
+ .build());
public static final ConfigOption<MemorySize> MAX_SIZE_LEVEL_BASE =
key("state.backend.forst.compaction.level.max-size-level-base")
.memoryType()
.defaultValue(MemorySize.parse("256mb"))
.withDescription(
- "The upper-bound of the total size of level base files in bytes. "
- + "The default value is '256MB'.");
+ Description.builder()
+ .text(
+ "The upper-bound of the total size of level base files in bytes. "
+ + "The default value is '%s'.",
+ text(MemorySize.parse("256mb").toString()))
+ .build());
public static final ConfigOption<MemorySize> WRITE_BUFFER_SIZE =
key("state.backend.forst.writebuffer.size")
@@ -289,15 +298,14 @@ public class ForStConfigurableOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "If true, every newly created SST file will contain a Bloom filter. "
- + "It is disabled by default.");
+ "Whether every newly created SST file will contain a Bloom filter. Default 'false'.");
public static final ConfigOption<Double> BLOOM_FILTER_BITS_PER_KEY =
key("state.backend.forst.bloom-filter.bits-per-key")
.doubleType()
.defaultValue(10.0)
.withDescription(
- "Bits per key that bloom filter will use, this only take effect when bloom filter is used. "
+ "Bits per key that the bloom filter will use, this only takes effect when the bloom filter is used. "
+ "The default value is 10.0.");
public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
@@ -305,7 +313,7 @@ public class ForStConfigurableOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "If true, ForSt will use block-based filter instead of full filter, this only take effect when bloom filter is used. "
+ "If set 'true', ForSt will use block-based filter instead of full filter, this only takes effect when bloom filter is used. "
+ "The default value is 'false'.");
public static final ConfigOption<Long> COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES =
@@ -325,14 +333,14 @@ public class ForStConfigurableOptions implements Serializable {
"Periodic compaction could speed up expired state entries cleanup, especially for state"
+ " entries rarely accessed. Files older than this value will be picked up for compaction,"
+ " and re-written to the same level as they were before. It makes sure a file goes through"
- + " compaction filters periodically. 0 means turning off periodic compaction.The default value is '30days'.");
+ + " compaction filters periodically. 0 means turning off periodic compaction.The default value is '30 d' (30 days).");
public static final ConfigOption<Double> RESTORE_OVERLAP_FRACTION_THRESHOLD =
key("state.backend.forst.restore-overlap-fraction-threshold")
.doubleType()
.defaultValue(0.0)
.withDescription(
- "The threshold of overlap fraction between the handle's key-group range and target key-group range. "
+ "The threshold of overlap fraction between the state handle's key-group range and target key-group range. "
+ "When restore base DB, only the handle which overlap fraction greater than or equal to threshold "
+ "has a chance to be an initial handle. "
+ "The default value is 0.0, there is always a handle will be selected for initialization. ");
@@ -351,7 +359,7 @@ public class ForStConfigurableOptions implements Serializable {
.defaultValue(Boolean.FALSE)
.withDescription(
"If true, during rescaling, the deleteFilesInRange API will be invoked "
- + "to clean up the useless files so that local disk space can be reclaimed more promptly.");
+ + "to clean up the useless key-values so that primary storage space can be reclaimed more promptly.");
static final ConfigOption<?>[] CANDIDATE_CONFIGS =
new ConfigOption<?>[] {
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
index 47603f505b8..103e700b460 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
@@ -43,8 +43,8 @@ import java.util.Set;
* <p>Statistics based metrics would report at the database level, it can return ticker or histogram
* kind results.
*
- * <p>Properties and doc comments are taken from RocksDB documentation. See <a
- * href="https://github.com/facebook/rocksdb/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
+ * <p>Properties and doc comments are taken from ForSt documentation. See <a
+ * href="https://github.com/ververica/ForSt/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
* db.h</a> for more information.
*/
@Experimental
@@ -52,36 +52,36 @@ public class ForStNativeMetricOptions implements Serializable {
private static final long serialVersionUID = 1L;
// --------------------------------------------------------------------------------------------
- // RocksDB property based metrics, report at column family level
+ // ForSt property based metrics, report at column family level
// --------------------------------------------------------------------------------------------
public static final String METRICS_COLUMN_FAMILY_AS_VARIABLE_KEY =
- "state.backend.rocksdb.metrics" + ".column-family-as-variable";
+ "state.backend.forst.metrics" + ".column-family-as-variable";
public static final ConfigOption<Boolean> MONITOR_NUM_IMMUTABLE_MEM_TABLES =
ConfigOptions.key(ForStProperty.NumImmutableMemTable.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of immutable memtables in RocksDB.");
+ .withDescription("Monitor the number of immutable memtables in ForSt.");
public static final ConfigOption<Boolean> MONITOR_MEM_TABLE_FLUSH_PENDING =
ConfigOptions.key(ForStProperty.MemTableFlushPending.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of pending memtable flushes in RocksDB.");
+ .withDescription("Monitor the number of pending memtable flushes in ForSt.");
public static final ConfigOption<Boolean> TRACK_COMPACTION_PENDING =
ConfigOptions.key(ForStProperty.CompactionPending.getConfigKey())
.booleanType()
.defaultValue(false)
.withDescription(
- "Track pending compactions in RocksDB. Returns 1 if a compaction is pending, 0 otherwise.");
+ "Track pending compactions in ForSt. Returns 1 if a compaction is pending, 0 otherwise.");
public static final ConfigOption<Boolean> MONITOR_BACKGROUND_ERRORS =
ConfigOptions.key(ForStProperty.BackgroundErrors.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of background errors in RocksDB.");
+ .withDescription("Monitor the number of background errors in ForSt.");
public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE =
ConfigOptions.key(ForStProperty.CurSizeActiveMemTable.getConfigKey())
@@ -137,7 +137,7 @@ public class ForStNativeMetricOptions implements Serializable {
ConfigOptions.key(ForStProperty.EstimateNumKeys.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Estimate the number of keys in RocksDB.");
+ .withDescription("Estimate the number of keys in ForSt.");
public static final ConfigOption<Boolean> ESTIMATE_TABLE_READERS_MEM =
ConfigOptions.key(ForStProperty.EstimateTableReadersMem.getConfigKey())
@@ -145,13 +145,13 @@ public class ForStNativeMetricOptions implements Serializable {
.defaultValue(false)
.withDescription(
"Estimate the memory used for reading SST tables, excluding memory"
- + " used in block cache (e.g.,filter and index blocks) in bytes.");
+ + " used in block cache (e.g. filter and index blocks) in bytes.");
public static final ConfigOption<Boolean> MONITOR_NUM_SNAPSHOTS =
ConfigOptions.key(ForStProperty.NumSnapshots.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of unreleased snapshots of the database.");
+ .withDescription("Monitor the number of unfinished snapshots of the ForSt.");
public static final ConfigOption<Boolean> MONITOR_NUM_LIVE_VERSIONS =
ConfigOptions.key(ForStProperty.NumLiveVersions.getConfigKey())
@@ -159,7 +159,7 @@ public class ForStNativeMetricOptions implements Serializable {
.defaultValue(false)
.withDescription(
"Monitor number of live versions. Version is an internal data structure. "
- + "See RocksDB file version_set.h for details. More live versions often mean more SST files are held "
+ + "See ForSt file version_set.h for details. More live versions often mean more SST files are held "
+ "from being deleted, by iterators or unfinished compactions.");
public static final ConfigOption<Boolean> ESTIMATE_LIVE_DATA_SIZE =
@@ -174,7 +174,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total size (bytes) of all SST files of all versions."
+ "Monitor the total size (bytes) of all SST files of all versions. "
+ "WARNING: may slow down online queries if there are too many files.");
public static final ConfigOption<Boolean> MONITOR_LIVE_SST_FILES_SIZE =
@@ -182,7 +182,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total size (bytes) of all SST files belonging to the latest version."
+ "Monitor the total size (bytes) of all SST files belonging to the latest version. "
+ "WARNING: may slow down online queries if there are too many files.");
public static final ConfigOption<Boolean> ESTIMATE_PENDING_COMPACTION_BYTES =
@@ -190,7 +190,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Estimated total number of bytes compaction needs to rewrite to get all levels "
+ "Estimated total number of bytes compaction needed to rewrite to get all levels "
+ "down to under target size. Not valid for other compactions than level-based.");
public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_COMPACTIONS =
@@ -217,7 +217,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Track whether write has been stopped in RocksDB. Returns 1 if write has been stopped, 0 otherwise.");
+ "Track whether write has been stopped in ForSt. The metric will return 1 if write has been stopped, 0 otherwise.");
public static final ConfigOption<Boolean> BLOCK_CACHE_CAPACITY =
ConfigOptions.key(ForStProperty.BlockCacheCapacity.getConfigKey())
@@ -244,91 +244,91 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether to expose the column family as a variable for RocksDB property based metrics.");
+ "Whether to expose the column family as a variable for ForSt property based metrics.");
public static final ConfigOption<Boolean> MONITOR_NUM_FILES_AT_LEVEL =
- ConfigOptions.key("state.backend.rocksdb.metrics.num-files-at-level")
+ ConfigOptions.key("state.backend.forst.metrics.num-files-at-level")
.booleanType()
.defaultValue(false)
.withDescription("Monitor the number of files at each level.");
// --------------------------------------------------------------------------------------------
- // RocksDB statistics based metrics, report at database level
+ // ForSt statistics based metrics, report at database level
// --------------------------------------------------------------------------------------------
public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_HIT =
- ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-hit")
+ ConfigOptions.key("state.backend.forst.metrics.block-cache-hit")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of block cache hit in RocksDB (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).");
+ "Monitor the total count of block cache hit in ForSt (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).");
public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_MISS =
- ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-miss")
+ ConfigOptions.key("state.backend.forst.metrics.block-cache-miss")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of block cache misses in RocksDB (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).");
+ "Monitor the total count of block cache misses in ForSt (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_USEFUL =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-useful")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-useful")
.booleanType()
.defaultValue(false)
.withDescription("Monitor the total count of reads avoided by bloom filter.");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_POSITIVE =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-positive")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-full-positive")
.booleanType()
.defaultValue(false)
.withDescription(
"Monitor the total count of reads not avoided by bloom full filter.");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_TRUE_POSITIVE =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-true-positive")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-full-true-positive")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of reads not avoided by bloom full filter and the data actually exists in RocksDB.");
+ "Monitor the total count of reads not avoided by bloom full filter and the data actually exists in ForSt.");
public static final ConfigOption<Boolean> MONITOR_BYTES_READ =
- ConfigOptions.key("state.backend.rocksdb.metrics.bytes-read")
+ ConfigOptions.key("state.backend.forst.metrics.bytes-read")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in RocksDB.");
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in ForSt.");
public static final ConfigOption<Boolean> MONITOR_ITER_BYTES_READ =
- ConfigOptions.key("state.backend.rocksdb.metrics.iter-bytes-read")
+ ConfigOptions.key("state.backend.forst.metrics.iter-bytes-read")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in RocksDB.");
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in ForSt.");
public static final ConfigOption<Boolean> MONITOR_BYTES_WRITTEN =
- ConfigOptions.key("state.backend.rocksdb.metrics.bytes-written")
+ ConfigOptions.key("state.backend.forst.metrics.bytes-written")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in RocksDB.");
+ "Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in ForSt.");
public static final ConfigOption<Boolean> MONITOR_COMPACTION_READ_BYTES =
- ConfigOptions.key("state.backend.rocksdb.metrics.compaction-read-bytes")
+ ConfigOptions.key("state.backend.forst.metrics.compaction-read-bytes")
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the bytes read during compaction in RocksDB.");
+ .withDescription("Monitor the bytes read during compaction in ForSt.");
public static final ConfigOption<Boolean> MONITOR_COMPACTION_WRITE_BYTES =
- ConfigOptions.key("state.backend.rocksdb.metrics.compaction-write-bytes")
+ ConfigOptions.key("state.backend.forst.metrics.compaction-write-bytes")
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the bytes written during compaction in RocksDB.");
+ .withDescription("Monitor the bytes written during compaction in ForSt.");
public static final ConfigOption<Boolean> MONITOR_STALL_MICROS =
- ConfigOptions.key("state.backend.rocksdb.metrics.stall-micros")
+ ConfigOptions.key("state.backend.forst.metrics.stall-micros")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the duration of writer requiring to wait for compaction or flush to finish in RocksDB.");
+ "Monitor the writer wait duration for compaction or flush to finish in ForSt.");
/** Creates a {@link ForStNativeMetricOptions} based on an external configuration. */
public static ForStNativeMetricOptions fromConfig(ReadableConfig config) {
@@ -666,14 +666,14 @@ public class ForStNativeMetricOptions implements Serializable {
}
/**
- * @return the enabled RocksDB property-based metrics
+ * @return the enabled ForSt property-based metrics
*/
public Collection<ForStProperty> getProperties() {
return Collections.unmodifiableCollection(properties);
}
/**
- * @return the enabled RocksDB statistics metrics.
+ * @return the enabled ForSt statistics metrics.
*/
public Collection<TickerType> getMonitorTickerTypes() {
return Collections.unmodifiableCollection(monitorTickerTypes);
@@ -689,7 +689,7 @@ public class ForStNativeMetricOptions implements Serializable {
}
/**
- * @return true if RocksDB statistics metrics are enabled, false otherwise.
+ * @return true if ForSt statistics metrics are enabled, false otherwise.
*/
public boolean isStatisticsEnabled() {
return !monitorTickerTypes.isEmpty();
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 73e43443332..8c38ed83038 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
@@ -35,6 +35,7 @@ import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHE
public class ForStOptions {
/** The local directory (on the TaskManager) where ForSt puts some meta files. */
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<String> LOCAL_DIRECTORIES =
ConfigOptions.key("state.backend.forst.local-dir")
.stringType()
@@ -42,7 +43,7 @@ public class ForStOptions {
.withDescription(
Description.builder()
.text(
- "The local directory (on the TaskManager) where ForSt puts some meta files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
+ "The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See %s for more details.",
TextElement.code(
ClusterOptions
.TASK_MANAGER_PROCESS_WORKING_DIR_BASE
@@ -50,6 +51,7 @@ public class ForStOptions {
.build());
/** The remote directory where ForSt puts its SST files. */
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<String> REMOTE_DIRECTORY =
ConfigOptions.key("state.backend.forst.remote-dir")
.stringType()
@@ -60,6 +62,7 @@ public class ForStOptions {
+ " Recognized shortcut name is '%s', which means that forst shares the directory with checkpoint.",
LOCAL_DIRECTORIES.key(), REMOTE_SHORTCUT_CHECKPOINT));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<String> CACHE_DIRECTORY =
ConfigOptions.key("state.backend.forst.cache.dir")
.stringType()
@@ -69,24 +72,29 @@ public class ForStOptions {
"The directory where ForSt caches its SST files, fallback to %s/cache if not configured.",
LOCAL_DIRECTORIES.key()));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Long> CACHE_SIZE_BASE_LIMIT =
ConfigOptions.key("state.backend.forst.cache.size-based-limit")
.longType()
.defaultValue(-1L)
.withDescription(
- "The size-based capacity limit of cache, a non-positive number indicates that there is no limit.");
+ "The size-based capacity limit of cache, a non-positive number indicates that there is no limit. "
+ + "The default value is '-1', which means that the cache size is not limited by size.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Long> CACHE_RESERVED_SIZE =
ConfigOptions.key("state.backend.forst.cache.reserve-size")
.longType()
.defaultValue(-1L)
.withDescription(
- "The reserved size of cache, when set to a positive number, dynamic space checking will be leveraged. "
- + "This option and the "
+ "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 '"
+ CACHE_SIZE_BASE_LIMIT.key()
- + "option can be set simultaneously, the smaller cache limit will be used as the upper limit.");
+ + "' option can be set simultaneously, the smaller cache limit will be used as the upper limit.");
/** The options factory class for ForSt to create DBOptions and ColumnFamilyOptions. */
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<String> OPTIONS_FACTORY =
ConfigOptions.key("state.backend.forst.options-factory")
.stringType()
@@ -96,37 +104,40 @@ public class ForStOptions {
+ "If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions "
+ "after loading ones from 'ForStConfigurableOptions' and pre-defined options.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<Boolean> USE_MANAGED_MEMORY =
ConfigOptions.key("state.backend.forst.memory.managed")
.booleanType()
.defaultValue(true)
.withDescription(
- "If set, the ForSt state backend will automatically configure itself to use the "
+ "If set true, the ForSt state backend will automatically configure itself to use the "
+ "managed memory budget of the task slot, and divide the memory over write buffers, indexes, "
- + "block caches, etc. That way, the three major uses of memory of ForSt will be capped.");
+ + "block caches, etc.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<MemorySize> FIX_PER_SLOT_MEMORY_SIZE =
ConfigOptions.key("state.backend.forst.memory.fixed-per-slot")
.memoryType()
.noDefaultValue()
.withDescription(
String.format(
- "The fixed total amount of memory, shared among all ForSt instances per slot. "
- + "This option overrides the '%s' option when configured.",
+ "The fixed total amount of memory per slot, shared among all ForSt instances."
+ + "This option overrides the '%s' option.",
USE_MANAGED_MEMORY.key()));
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<MemorySize> FIX_PER_TM_MEMORY_SIZE =
ConfigOptions.key("state.backend.forst.memory.fixed-per-tm")
.memoryType()
.noDefaultValue()
.withDescription(
String.format(
- "The fixed total amount of memory, shared among all ForSt instances per Task Manager (cluster-level option). "
- + "This option only takes effect if neither '%s' nor '%s' are not configured. If none is configured "
+ "The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. "
+ + "This option only takes effect if '%s' is set to false and '%s' is not configured. If so, "
+ "then each ForSt column family state has its own memory caches (as controlled by the column "
+ "family options). "
- + "The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the same level (config.yaml)."
- + "Note, that this feature breaks resource isolation between the slots",
+ + "The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the same level (config.yaml). "
+ + "Note that this feature breaks resource isolation between the slots.",
USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
public static final ConfigOption<Double> WRITE_BUFFER_RATIO =
@@ -149,10 +160,11 @@ public class ForStOptions {
+ "compression dictionary blocks. This option only has an effect when '%s' or '%s' are configured.",
USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Boolean> USE_PARTITIONED_INDEX_FILTERS =
ConfigOptions.key("state.backend.forst.memory.partitioned-index-filters")
.booleanType()
- .defaultValue(false)
+ .defaultValue(true)
.withDescription(
String.format(
"With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with "
@@ -163,7 +175,7 @@ public class ForStOptions {
USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
/** Choice of timer service implementation. */
- @Documentation.Section(Documentation.Sections.STATE_BACKEND_ROCKSDB)
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<ForStStateBackend.PriorityQueueStateType>
TIMER_SERVICE_FACTORY =
ConfigOptions.key("state.backend.forst.timer-service.factory")
@@ -172,33 +184,42 @@ public class ForStOptions {
.withDescription(
"This determines the factory for timer service state implementation.");
- /** The cache size per key-group for ROCKSDB timer service factory implementation. */
- @Documentation.Section(Documentation.Sections.STATE_BACKEND_ROCKSDB)
+ /** The cache size per key-group for ForSt timer service factory implementation. */
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> FORST_TIMER_SERVICE_FACTORY_CACHE_SIZE =
ConfigOptions.key("state.backend.forst.timer-service.cache-size")
.intType()
.defaultValue(128)
.withDescription(
String.format(
- "The cache size per keyGroup of rocksdb timer service factory. This option only has an effect "
+ "The cache size per keyGroup of ForSt timer service factory. This option only has an effect "
+ "when '%s' is configured to '%s'. Increasing this value can improve the performance "
- + "of rocksdb timer service, but consumes more heap memory at the same time.",
+ + "of ForSt timer service, but consumes more heap memory at the same time.",
TIMER_SERVICE_FACTORY.key(), ForStDB.name()));
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<Boolean> EXECUTOR_COORDINATOR_INLINE =
ConfigOptions.key("state.backend.forst.executor.inline-coordinator")
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether to let the task thread be the coordinator thread responsible for distributing requests.");
+ "Whether to let the task thread be the coordinator thread responsible for distributing requests. "
+ + "If set to 'true', the task thread will be responsible for distributing requests, "
+ + "otherwise, a dedicated coordinator thread will be used. "
+ + "The default value is 'false'.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<Boolean> EXECUTOR_WRITE_IO_INLINE =
ConfigOptions.key("state.backend.forst.executor.inline-write")
.booleanType()
.defaultValue(true)
.withDescription(
- "Whether to let write requests executed within the coordinator thread.");
+ "Whether to let write requests be executed within the coordinator thread. "
+ + "If set to 'true', write requests will be executed within the coordinator thread, "
+ + "otherwise, a dedicated write thread will be used. "
+ + "The default value is 'true'.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> EXECUTOR_READ_IO_PARALLELISM =
ConfigOptions.key("state.backend.forst.executor.read-io-parallelism")
.intType()
@@ -206,6 +227,7 @@ public class ForStOptions {
.withDescription(
"The number of threads used for read IO operations in the executor.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> EXECUTOR_WRITE_IO_PARALLELISM =
ConfigOptions.key("state.backend.forst.executor.write-io-parallelism")
.intType()