[FLINK-36439][docs] Documents for ForSt State Backend

Author: Zakelly

Diff for: 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 for: 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 for: 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 for: 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 for: 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 for: 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,18 +188,29 @@ 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]({{< ref "docs/deployment/config#flink-configuration-file" >}}).
 {{< /hint >}}
 
 
 ### Setting Default State Backend
 
 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 for: 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 &lt;WORKING_DIR&gt;/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 same level (config.yaml). Note that this feature breaks resource isolation between the slots.</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.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>