This is an automated email from the ASF dual-hosted git repository. hangxiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 38070a1bde8b3280ba852d7956a7d8e5c9fb9dd4 Author: Hangxiang Yu <[email protected]> AuthorDate: Mon Jun 3 10:40:40 2024 +0800 [FLINK-34482][checkpoint] Rename checkpointing options --- docs/content.zh/docs/deployment/cli.md | 6 +- docs/content.zh/docs/deployment/config.md | 4 +- .../docs/dev/python/table/table_environment.md | 2 +- docs/content.zh/docs/dev/table/sql/job.md | 4 +- docs/content.zh/docs/dev/table/sqlClient.md | 2 +- docs/content.zh/docs/ops/state/checkpoints.md | 4 +- docs/content.zh/docs/ops/state/savepoints.md | 4 +- docs/content.zh/docs/ops/state/state_backends.md | 28 ++-- .../docs/try-flink/flink-operations-playground.md | 2 +- docs/content/docs/deployment/cli.md | 6 +- docs/content/docs/deployment/config.md | 4 +- .../docs/dev/python/table/table_environment.md | 2 +- docs/content/docs/dev/table/sql/job.md | 4 +- docs/content/docs/dev/table/sqlClient.md | 2 +- docs/content/docs/ops/state/checkpoints.md | 4 +- docs/content/docs/ops/state/savepoints.md | 4 +- docs/content/docs/ops/state/state_backends.md | 28 ++-- .../docs/try-flink/flink-operations-playground.md | 2 +- .../generated/checkpoint_file_merging_section.html | 12 +- .../generated/checkpointing_configuration.html | 176 ++++++++++----------- .../generated/common_state_backends_section.html | 38 ++--- .../generated/expert_state_backends_section.html | 10 +- .../flink/configuration/CheckpointingOptions.java | 70 ++++---- .../test-scripts/test_file_sink.sh | 2 +- flink-python/pyflink/datastream/state_backend.py | 2 +- .../checkpoint/CheckpointSubsumeHelper.java | 4 +- .../runtime/state/CheckpointStorageLoader.java | 3 +- .../runtime/state/memory/MemoryStateBackend.java | 2 +- .../job/savepoints/SavepointHandlersTest.java | 2 +- .../savepoints/StopWithSavepointHandlersTest.java | 2 +- .../service/MaterializedTableStatementITCase.java | 3 +- 31 files changed, 226 insertions(+), 212 deletions(-) diff --git a/docs/content.zh/docs/deployment/cli.md b/docs/content.zh/docs/deployment/cli.md index 7acb433dcee..88b26b64198 100644 --- a/docs/content.zh/docs/deployment/cli.md +++ b/docs/content.zh/docs/deployment/cli.md @@ -119,7 +119,7 @@ Savepoint completed. Path: file:/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0 You can resume your program from this savepoint with the run command. ``` The savepoint folder is optional and needs to be specified if -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. Lastly, you can optionally provide what should be the [binary format]({{< ref "docs/ops/state/savepoints" >}}#savepoint-format) of the savepoint. @@ -238,7 +238,7 @@ Suspending job "cca7bc1061d61cf15238e92312c2fc20" with a savepoint. Savepoint completed. Path: file:/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab ``` We have to use `--savepointPath` to specify the savepoint folder if -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. If the `--drain` flag is specified, then a `MAX_WATERMARK` will be emitted before the last checkpoint barrier. This will make all registered event-time timers fire, thus flushing out any state that @@ -365,7 +365,7 @@ Here's an overview of actions supported by Flink's CLI tool: <td> This action can be used to create or disposing savepoints for a given job. It might be necessary to specify a savepoint directory besides the JobID, if the - <a href="{{< ref "docs/deployment/config" >}}#state-savepoints-dir">state.savepoints.dir</a> + <a href="{{< ref "docs/deployment/config" >}}#state-savepoints-dir">execution.checkpointing.savepoint-dir</a> parameter was not specified in <code class="highlighter-rouge">Flink configuration file</code>. </td> </tr> diff --git a/docs/content.zh/docs/deployment/config.md b/docs/content.zh/docs/deployment/config.md index 0ef52009d5b..896ea3102ca 100644 --- a/docs/content.zh/docs/deployment/config.md +++ b/docs/content.zh/docs/deployment/config.md @@ -161,8 +161,8 @@ 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.checkpoints.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*. - - `state.savepoints.dir`: The default directory for savepoints. Takes a path URI, similar to `state.checkpoints.dir`. + - `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. **Web UI** diff --git a/docs/content.zh/docs/dev/python/table/table_environment.md b/docs/content.zh/docs/dev/python/table/table_environment.md index 8b59e97cb7e..9d3bb72fffb 100644 --- a/docs/content.zh/docs/dev/python/table/table_environment.md +++ b/docs/content.zh/docs/dev/python/table/table_environment.md @@ -835,5 +835,5 @@ table_env.get_config().set("execution.checkpointing.interval", "3min") table_env.get_config().set("state.backend.type", "rocksdb") # 设置 RocksDB statebackend 所需要的 checkpoint 目录 -table_env.get_config().set("state.checkpoints.dir", "file:///tmp/checkpoints/") +table_env.get_config().set("execution.checkpointing.dir", "file:///tmp/checkpoints/") ``` diff --git a/docs/content.zh/docs/dev/table/sql/job.md b/docs/content.zh/docs/dev/table/sql/job.md index 7417f7827fe..5661969f4f2 100644 --- a/docs/content.zh/docs/dev/table/sql/job.md +++ b/docs/content.zh/docs/dev/table/sql/job.md @@ -52,7 +52,7 @@ Flink SQL> SHOW JOBS; | 228d70913eab60dda85c5e7f78b5782c | myjob | RUNNING | 2023-02-11T05:03:51.523 | +----------------------------------+----------+---------+-------------------------+ -Flink SQL> SET 'state.savepoints.dir'='file:/tmp/'; +Flink SQL> SET 'execution.checkpointing.savepoint-dir'='file:/tmp/'; [INFO] Execute statement succeeded. Flink SQL> STOP JOB '228d70913eab60dda85c5e7f78b5782c' WITH SAVEPOINT; @@ -85,7 +85,7 @@ STOP JOB '<job_id>' [WITH SAVEPOINT] [WITH DRAIN] **WITH SAVEPOINT** 在作业停止之前执行 Savepoin。 Savepoint 的路径可以通过集群配置的 -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) 指定， +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) 指定， 或者通过 `SET` 语句指定（后者有更高优先级）。 **WITH DRAIN** diff --git a/docs/content.zh/docs/dev/table/sqlClient.md b/docs/content.zh/docs/dev/table/sqlClient.md index c29d30a0228..32fb4fe215b 100644 --- a/docs/content.zh/docs/dev/table/sqlClient.md +++ b/docs/content.zh/docs/dev/table/sqlClient.md @@ -927,7 +927,7 @@ Flink SQL> STOP JOB '228d70913eab60dda85c5e7f78b5782c' WITH SAVEPOINT; +-----------------------------------------+ ``` -The savepoint path could be specified with [state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) +The savepoint path could be specified with [execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) either in the cluster configuration or session configuration (the latter would take precedence). For more details about stopping jobs, please refer to [Job Statements]({{< ref "docs/dev/table/sql/job" >}}#stop-job). diff --git a/docs/content.zh/docs/ops/state/checkpoints.md b/docs/content.zh/docs/ops/state/checkpoints.md index e45e9077af1..70f2abe1927 100644 --- a/docs/content.zh/docs/ops/state/checkpoints.md +++ b/docs/content.zh/docs/ops/state/checkpoints.md @@ -100,7 +100,7 @@ config.setExternalizedCheckpointRetention(ExternalizedCheckpointRetention.RETAIN ### 目录结构 -与 [savepoints]({{< ref "docs/ops/state/savepoints" >}}) 相似，checkpoint 由元数据文件、数据文件（与 state backend 相关）组成。可通过配置文件中 "state.checkpoints.dir" 配置项来指定元数据文件和数据文件的存储路径，另外也可以在代码中针对单个作业特别指定该配置项。 +与 [savepoints]({{< ref "docs/ops/state/savepoints" >}}) 相似，checkpoint 由元数据文件、数据文件（与 state backend 相关）组成。可通过配置文件中 "execution.checkpointing.dir" 配置项来指定元数据文件和数据文件的存储路径，另外也可以在代码中针对单个作业特别指定该配置项。 当前的 checkpoint 目录结构（由 [FLINK-8531](https://issues.apache.org/jira/browse/FLINK-8531) 引入）如下所示: @@ -125,7 +125,7 @@ config.setExternalizedCheckpointRetention(ExternalizedCheckpointRetention.RETAIN #### 通过配置文件全局配置 ```yaml -state.checkpoints.dir: hdfs:///checkpoints/ +execution.checkpointing.dir: hdfs:///checkpoints/ ``` #### 创建 state backend 对单个作业进行配置 diff --git a/docs/content.zh/docs/ops/state/savepoints.md b/docs/content.zh/docs/ops/state/savepoints.md index 25ae077a88a..dcd03bf868f 100644 --- a/docs/content.zh/docs/ops/state/savepoints.md +++ b/docs/content.zh/docs/ops/state/savepoints.md @@ -255,12 +255,12 @@ $ bin/flink savepoint -d :savepointPath ### 配置 -你可以通过 `state.savepoints.dir` 配置 savepoint 的默认目录。 触发 savepoint 时，将使用此目录来存储 savepoint。 你可以通过使用触发器命令指定自定义目标目录来覆盖缺省值（请参阅[`:targetDirectory`参数](#触发-savepoint-1)）。 +你可以通过 `execution.checkpointing.savepoint-dir` 配置 savepoint 的默认目录。 触发 savepoint 时，将使用此目录来存储 savepoint。 你可以通过使用触发器命令指定自定义目标目录来覆盖缺省值（请参阅[`:targetDirectory`参数](#触发-savepoint-1)）。 ```yaml # 默认 Savepoint 目标目录 -state.savepoints.dir: hdfs:///flink/savepoints +execution.checkpointing.savepoint-dir: hdfs:///flink/savepoints ``` 如果既未配置缺省值也未指定自定义目标目录，则触发 Savepoint 将失败。 diff --git a/docs/content.zh/docs/ops/state/state_backends.md b/docs/content.zh/docs/ops/state/state_backends.md index a8aebf9419e..dcdc22e11d3 100644 --- a/docs/content.zh/docs/ops/state/state_backends.md +++ b/docs/content.zh/docs/ops/state/state_backends.md @@ -171,7 +171,7 @@ env = StreamExecutionEnvironment.get_execution_environment(config) 或使用实现了 state backend 工厂 {{< gh_link file="flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendFactory.java" name="StateBackendFactory" >}} 的类的全限定类名， 例如： EmbeddedRocksDBStateBackend 对应为 `org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackendFactory`。 -`state.checkpoints.dir` 选项指定了所有 State Backend 写 CheckPoint 数据和写元数据文件的目录。 +`execution.checkpointing.dir` 选项指定了所有 State Backend 写 CheckPoint 数据和写元数据文件的目录。 你能在 [这里]({{< ref "docs/ops/state/checkpoints" >}}#directory-structure) 找到关于 CheckPoint 目录结构的详细信息。 配置文件的部分示例如下所示： @@ -184,7 +184,7 @@ state.backend: hashmap # 存储快照的目录 -state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints +execution.checkpointing.dir: hdfs://namenode:40010/flink/checkpoints ``` <a name="rocksdb-state-backend-details"></a> @@ -204,7 +204,7 @@ RocksDB 支持*增量快照*。不同于产生一个包含所有数据的全量 和基于全量快照的恢复时间相比，如果网络带宽是瓶颈，那么基于增量快照恢复可能会消耗更多时间，因为增量快照包含的 sst 文件之间可能存在数据重叠导致需要下载的数据量变大；而当 CPU 或者 IO 是瓶颈的时候，基于增量快照恢复会更快，因为从增量快照恢复不需要解析 Flink 的统一快照格式来重建本地的 RocksDB 数据表，而是可以直接基于 sst 文件加载。 虽然状态数据量很大时我们推荐使用增量快照，但这并不是默认的快照机制，您需要通过下述配置手动开启该功能： - - 在 [Flink 配置文件]({{< ref "docs/deployment/config#flink-配置文件" >}}) 中设置：`state.backend.incremental: true` 或者 + - 在 [Flink 配置文件]({{< ref "docs/deployment/config#flink-配置文件" >}}) 中设置：`execution.checkpointing.incremental: true` 或者 - 在代码中按照右侧方式配置（来覆盖默认配置）：`EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);` 需要注意的是，一旦启用了增量快照，网页上展示的 `Checkpointed Data Size` 只代表增量上传的数据量，而不是一次快照的完整数据量。 @@ -403,7 +403,7 @@ Changelog 是一项旨在减少 checkpointing 时间的功能，因此也可以 ```yaml state.changelog.enabled: true state.changelog.storage: filesystem # 当前只支持 filesystem 和 memory（仅供测试用） -state.changelog.dstl.dfs.base-path: s3://<bucket-name> # 类似于 state.checkpoints.dir +state.changelog.dstl.dfs.base-path: s3://<bucket-name> # 类似于 execution.checkpointing.dir ``` 请将如下配置保持默认值 （参见[限制](#limitations)）: @@ -492,7 +492,7 @@ state.backend: hashmap # Optional, Flink will automatically default to JobManagerCheckpointStorage # when no checkpoint directory is specified. -state.checkpoint-storage: jobmanager +execution.checkpointing.storage: jobmanager ``` #### 代码配置 @@ -517,7 +517,7 @@ env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage) ```python config = Configuration() config.set_string('state.backend.type', 'hashmap') -config.set_string('state.checkpoint-storage', 'jobmanager') +config.set_string('execution.checkpointing.storage', 'jobmanager') env = StreamExecutionEnvironment.get_execution_environment(config) ``` {{< /tab >}} @@ -531,11 +531,11 @@ env = StreamExecutionEnvironment.get_execution_environment(config) ```yaml state.backend: hashmap -state.checkpoints.dir: file:///checkpoint-dir/ +execution.checkpointing.dir: file:///checkpoint-dir/ # Optional, Flink will automatically default to FileSystemCheckpointStorage # when a checkpoint directory is specified. -state.checkpoint-storage: filesystem +execution.checkpointing.storage: filesystem ``` #### 代码配置 @@ -572,8 +572,8 @@ env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(" ```python config = Configuration() config.set_string('state.backend.type', 'hashmap') -config.set_string('state.checkpoint-storage', 'filesystem') -config.set_string('state.checkpoints.dir', 'file:///checkpoint-dir') +config.set_string('execution.checkpointing.storage', 'filesystem') +config.set_string('execution.checkpointing.dir', 'file:///checkpoint-dir') env = StreamExecutionEnvironment.get_execution_environment(config) @@ -593,11 +593,11 @@ env.configure(config); ```yaml state.backend: rocksdb -state.checkpoints.dir: file:///checkpoint-dir/ +execution.checkpointing.dir: file:///checkpoint-dir/ # Optional, Flink will automatically default to FileSystemCheckpointStorage # when a checkpoint directory is specified. -state.checkpoint-storage: filesystem +execution.checkpointing.storage: filesystem ``` #### 代码配置 @@ -636,8 +636,8 @@ env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(" ```python config = Configuration() config.set_string('state.backend.type', 'hashmap') -config.set_string('state.checkpoint-storage', 'filesystem') -config.set_string('state.checkpoints.dir', 'file:///checkpoint-dir') +config.set_string('execution.checkpointing.storage', 'filesystem') +config.set_string('execution.checkpointing.dir', 'file:///checkpoint-dir') env = StreamExecutionEnvironment.get_execution_environment(config) diff --git a/docs/content.zh/docs/try-flink/flink-operations-playground.md b/docs/content.zh/docs/try-flink/flink-operations-playground.md index 199831654f8..b073453ab49 100644 --- a/docs/content.zh/docs/try-flink/flink-operations-playground.md +++ b/docs/content.zh/docs/try-flink/flink-operations-playground.md @@ -377,7 +377,7 @@ Suspending job "<job-id>" with a savepoint. Suspended job "<job-id>" with a savepoint. ``` -Savepoint 已保存在 `state.savepoints.dir` 指定的路径中，该配置在 [*Flink 配置文件*]({{< ref "docs/deployment/config#flink-配置文件" >}}) +Savepoint 已保存在 `execution.checkpointing.savepoint-dir` 指定的路径中，该配置在 [*Flink 配置文件*]({{< ref "docs/deployment/config#flink-配置文件" >}}) 中定义，[*Flink 配置文件*]({{< ref "docs/deployment/config#flink-配置文件" >}}) 挂载在本机的 */tmp/flink-savepoints-directory/* 目录下。 在下一步操作中我们会用到这个 Savepoint 路径，如果我们是通过 REST API 操作的， 那么 Savepoint 路径会随着响应结果一起返回，我们可以直接查看文件系统来确认 Savepoint 保存情况。 diff --git a/docs/content/docs/deployment/cli.md b/docs/content/docs/deployment/cli.md index 80ca0460497..3009a682895 100644 --- a/docs/content/docs/deployment/cli.md +++ b/docs/content/docs/deployment/cli.md @@ -117,7 +117,7 @@ Savepoint completed. Path: file:/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0 You can resume your program from this savepoint with the run command. ``` The savepoint folder is optional and needs to be specified if -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. Lastly, you can optionally provide what should be the [binary format]({{< ref "docs/ops/state/savepoints" >}}#savepoint-format) of the savepoint. @@ -236,7 +236,7 @@ Suspending job "cca7bc1061d61cf15238e92312c2fc20" with a savepoint. Savepoint completed. Path: file:/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab ``` We have to use `--savepointPath` to specify the savepoint folder if -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) isn't set. If the `--drain` flag is specified, then a `MAX_WATERMARK` will be emitted before the last checkpoint barrier. This will make all registered event-time timers fire, thus flushing out any state that @@ -363,7 +363,7 @@ Here's an overview of actions supported by Flink's CLI tool: <td> This action can be used to create or disposing savepoints for a given job. It might be necessary to specify a savepoint directory besides the JobID, if the - <a href="{{< ref "docs/deployment/config" >}}#state-savepoints-dir">state.savepoints.dir</a> + <a href="{{< ref "docs/deployment/config" >}}#state-savepoints-dir">execution.checkpointing.savepoint-dir</a> parameter was not specified in <code class="highlighter-rouge">Flink configuration file</code>. </td> </tr> diff --git a/docs/content/docs/deployment/config.md b/docs/content/docs/deployment/config.md index d856dcab109..ada52f3b63f 100644 --- a/docs/content/docs/deployment/config.md +++ b/docs/content/docs/deployment/config.md @@ -161,8 +161,8 @@ 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.checkpoints.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*. - - `state.savepoints.dir`: The default directory for savepoints. Takes a path URI, similar to `state.checkpoints.dir`. + - `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. **Web UI** diff --git a/docs/content/docs/dev/python/table/table_environment.md b/docs/content/docs/dev/python/table/table_environment.md index b9dd7583688..42d6462bb75 100644 --- a/docs/content/docs/dev/python/table/table_environment.md +++ b/docs/content/docs/dev/python/table/table_environment.md @@ -839,5 +839,5 @@ table_env.get_config().set("execution.checkpointing.interval", "3min") table_env.get_config().set("state.backend.type", "rocksdb") # set the checkpoint directory, which is required by the RocksDB statebackend -table_env.get_config().set("state.checkpoints.dir", "file:///tmp/checkpoints/") +table_env.get_config().set("execution.checkpointing.dir", "file:///tmp/checkpoints/") ``` diff --git a/docs/content/docs/dev/table/sql/job.md b/docs/content/docs/dev/table/sql/job.md index b00fc0fa9cf..e9d00f4f2a2 100644 --- a/docs/content/docs/dev/table/sql/job.md +++ b/docs/content/docs/dev/table/sql/job.md @@ -52,7 +52,7 @@ Flink SQL> SHOW JOBS; | 228d70913eab60dda85c5e7f78b5782c | myjob | RUNNING | 2023-02-11T05:03:51.523 | +----------------------------------+----------+---------+-------------------------+ -Flink SQL> SET 'state.savepoints.dir'='file:/tmp/'; +Flink SQL> SET 'execution.checkpointing.savepoint-dir'='file:/tmp/'; [INFO] Execute statement succeeded. Flink SQL> STOP JOB '228d70913eab60dda85c5e7f78b5782c' WITH SAVEPOINT; @@ -85,7 +85,7 @@ Stop the specified job. **WITH SAVEPOINT** Perform a savepoint right before stopping the job. The savepoint path could be specified with -[state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) either in +[execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) either in the cluster configuration or via `SET` statements (the latter would take precedence). **WITH DRAIN** diff --git a/docs/content/docs/dev/table/sqlClient.md b/docs/content/docs/dev/table/sqlClient.md index a3ba08084b2..40255f47ef3 100644 --- a/docs/content/docs/dev/table/sqlClient.md +++ b/docs/content/docs/dev/table/sqlClient.md @@ -865,7 +865,7 @@ Flink SQL> STOP JOB '228d70913eab60dda85c5e7f78b5782c' WITH SAVEPOINT; +-----------------------------------------+ ``` -The savepoint path could be specified with [state.savepoints.dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) +The savepoint path could be specified with [execution.checkpointing.savepoint-dir]({{< ref "docs/deployment/config" >}}#state-savepoints-dir) either in the cluster configuration or session configuration (the latter would take precedence). For more details about stopping jobs, please refer to [Job Statements]({{< ref "docs/dev/table/sql/job" >}}#stop-job). diff --git a/docs/content/docs/ops/state/checkpoints.md b/docs/content/docs/ops/state/checkpoints.md index db6a8b186b4..35a3d7187f1 100644 --- a/docs/content/docs/ops/state/checkpoints.md +++ b/docs/content/docs/ops/state/checkpoints.md @@ -112,7 +112,7 @@ The `ExternalizedCheckpointRetention` mode configures what happens with checkpoi Similarly to [savepoints]({{< ref "docs/ops/state/savepoints" >}}), a checkpoint consists of a meta data file and, depending on the state backend, some additional data files. The meta data file and data files are stored in the directory that is -configured via `state.checkpoints.dir` in the configuration files, +configured via `execution.checkpointing.dir` in the configuration files, and also can be specified for per job in the code. The current checkpoint directory layout ([introduced by FLINK-8531](https://issues.apache.org/jira/browse/FLINK-8531)) is as follows: @@ -138,7 +138,7 @@ The checkpoint directory is not part of a public API and can be changed in the f #### Configure globally via configuration files ```yaml -state.checkpoints.dir: hdfs:///checkpoints/ +execution.checkpointing.dir: hdfs:///checkpoints/ ``` #### Configure for per job on the checkpoint configuration diff --git a/docs/content/docs/ops/state/savepoints.md b/docs/content/docs/ops/state/savepoints.md index cf08141bad8..e7544c76466 100644 --- a/docs/content/docs/ops/state/savepoints.md +++ b/docs/content/docs/ops/state/savepoints.md @@ -306,13 +306,13 @@ Note that it is possible to also manually delete a savepoint via regular file sy ### Configuration -You can configure a default savepoint target directory via the `state.savepoints.dir` key or `StreamExecutionEnvironment`. When triggering savepoints, this directory will be used to store the savepoint. You can overwrite the default by specifying a custom target directory with the trigger commands (see the [`:targetDirectory` argument](#trigger-a-savepoint)). +You can configure a default savepoint target directory via the `execution.checkpointing.savepoint-dir` key or `StreamExecutionEnvironment`. When triggering savepoints, this directory will be used to store the savepoint. You can overwrite the default by specifying a custom target directory with the trigger commands (see the [`:targetDirectory` argument](#trigger-a-savepoint)). {{< tabs "config" >}} {{< tab "config.yaml" >}} ```yaml # Default savepoint target directory -state.savepoints.dir: hdfs:///flink/savepoints +execution.checkpointing.savepoint-dir: hdfs:///flink/savepoints ``` {{< /tab >}} {{< tab "Java" >}} diff --git a/docs/content/docs/ops/state/state_backends.md b/docs/content/docs/ops/state/state_backends.md index d2770eb1bcd..615eb96a879 100644 --- a/docs/content/docs/ops/state/state_backends.md +++ b/docs/content/docs/ops/state/state_backends.md @@ -163,7 +163,7 @@ Possible values for the config entry are *hashmap* (HashMapStateBackend), *rocks 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.contrib.streaming.state.EmbeddedRocksDBStateBackendFactory` for EmbeddedRocksDBStateBackend. -The `state.checkpoints.dir` option defines the directory to which all backends write checkpoint data and meta data files. +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). A sample section in the configuration file could look as follows: @@ -173,7 +173,7 @@ A sample section in the configuration file could look as follows: state.backend: hashmap # Directory for storing checkpoints -state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints +execution.checkpointing.dir: hdfs://namenode:40010/flink/checkpoints ``` ## RocksDB State Backend Details @@ -190,7 +190,7 @@ An incremental checkpoint builds upon (typically multiple) previous checkpoints. Recovery time of incremental checkpoints may be longer or shorter compared to full checkpoints. If your network bandwidth is the bottleneck, it may take a bit longer to restore from an incremental checkpoint, because it implies fetching more data (more deltas). Restoring from an incremental checkpoint is faster, if the bottleneck is your CPU or IOPs, because restoring from an incremental checkpoint means not re-building the local RocksDB tables from Flink's canonical key/value snapshot f [...] While we encourage the use of incremental checkpoints for large state, you need to enable this feature manually: - - Setting a default in your [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}): `state.backend.incremental: true` will enable incremental checkpoints, unless the application overrides this setting in the code. + - Setting a default in your [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}): `execution.checkpointing.incremental: true` will enable incremental checkpoints, unless the application overrides this setting in the code. - You can alternatively configure this directly in the code (overrides the config default): `EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);` Notice that once incremental checkpoont is enabled, the `Checkpointed Data Size` showed in web UI only represents the @@ -404,7 +404,7 @@ Here is an example configuration in YAML: ```yaml state.changelog.enabled: true state.changelog.storage: filesystem # currently, only filesystem and memory (for tests) are supported -state.changelog.dstl.dfs.base-path: s3://<bucket-name> # similar to state.checkpoints.dir +state.changelog.dstl.dfs.base-path: s3://<bucket-name> # similar to execution.checkpointing.dir ``` Please keep the following defaults (see [limitations](#limitations)): @@ -482,7 +482,7 @@ state.backend: hashmap # Optional, Flink will automatically default to JobManagerCheckpointStorage # when no checkpoint directory is specified. -state.checkpoint-storage: jobmanager +execution.checkpointing.storage: jobmanager ``` #### Code Configuration @@ -507,7 +507,7 @@ env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage) ```python config = Configuration() config.set_string('state.backend.type', 'hashmap') -config.set_string('state.checkpoint-storage', 'jobmanager') +config.set_string('execution.checkpointing.storage', 'jobmanager') env = StreamExecutionEnvironment.get_execution_environment(config) ``` {{< /tab >}} @@ -521,11 +521,11 @@ The legacy `FsStateBackend` is equivalent to using [`HashMapStateBackend`](#the- ```yaml state.backend: hashmap -state.checkpoints.dir: file:///checkpoint-dir/ +execution.checkpointing.dir: file:///checkpoint-dir/ # Optional, Flink will automatically default to FileSystemCheckpointStorage # when a checkpoint directory is specified. -state.checkpoint-storage: filesystem +execution.checkpointing.storage: filesystem ``` #### Code Configuration @@ -562,8 +562,8 @@ env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(" ```python config = Configuration() config.set_string('state.backend.type', 'hashmap') -config.set_string('state.checkpoint-storage', 'filesystem') -config.set_string('state.checkpoints.dir', 'file:///checkpoint-dir') +config.set_string('execution.checkpointing.storage', 'filesystem') +config.set_string('execution.checkpointing.dir', 'file:///checkpoint-dir') env = StreamExecutionEnvironment.get_execution_environment(config) @@ -583,11 +583,11 @@ The legacy `RocksDBStateBackend` is equivalent to using [`EmbeddedRocksDBStateBa ```yaml state.backend: rocksdb -state.checkpoints.dir: file:///checkpoint-dir/ +execution.checkpointing.dir: file:///checkpoint-dir/ # Optional, Flink will automatically default to FileSystemCheckpointStorage # when a checkpoint directory is specified. -state.checkpoint-storage: filesystem +execution.checkpointing.storage: filesystem ``` #### Code Configuration @@ -626,8 +626,8 @@ env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(" ```python config = Configuration() config.set_string('state.backend.type', 'rocksdb') -config.set_string('state.checkpoint-storage', 'filesystem') -config.set_string('state.checkpoints.dir', 'file:///checkpoint-dir') +config.set_string('execution.checkpointing.storage', 'filesystem') +config.set_string('execution.checkpointing.dir', 'file:///checkpoint-dir') env = StreamExecutionEnvironment.get_execution_environment(config) diff --git a/docs/content/docs/try-flink/flink-operations-playground.md b/docs/content/docs/try-flink/flink-operations-playground.md index cd93943c9ad..2e1ae96820a 100644 --- a/docs/content/docs/try-flink/flink-operations-playground.md +++ b/docs/content/docs/try-flink/flink-operations-playground.md @@ -367,7 +367,7 @@ Suspending job "<job-id>" with a savepoint. Savepoint completed. Path: file:<savepoint-path> ``` -The Savepoint has been stored to the `state.savepoints.dir` configured in the [*Flink configuration file*]({{< ref "docs/deployment/config#flink-configuration-file" >}}), +The Savepoint has been stored to the `execution.checkpointing.savepoint-dir` configured in the [*Flink configuration file*]({{< ref "docs/deployment/config#flink-configuration-file" >}}), which is mounted under */tmp/flink-savepoints-directory/* on your local machine. You will need the path to this Savepoint in the next step. diff --git a/docs/layouts/shortcodes/generated/checkpoint_file_merging_section.html b/docs/layouts/shortcodes/generated/checkpoint_file_merging_section.html index f67c75f1744..b8e09f77dc5 100644 --- a/docs/layouts/shortcodes/generated/checkpoint_file_merging_section.html +++ b/docs/layouts/shortcodes/generated/checkpoint_file_merging_section.html @@ -9,31 +9,31 @@ </thead> <tbody> <tr> - <td><h5>state.checkpoints.file-merging.enabled</h5></td> + <td><h5>execution.checkpointing.file-merging.enabled</h5></td> <td style="word-wrap: break-word;">false</td> <td>Boolean</td> <td>Whether to enable merging multiple checkpoint files into one, which will greatly reduce the number of small checkpoint files. This is an experimental feature under evaluation, make sure you're aware of the possible effects of enabling it.</td> </tr> <tr> - <td><h5>state.checkpoints.file-merging.across-checkpoint-boundary</h5></td> + <td><h5>execution.checkpointing.file-merging.across-checkpoint-boundary</h5></td> <td style="word-wrap: break-word;">false</td> <td>Boolean</td> - <td>Only relevant if <code class="highlighter-rouge">state.checkpoints.file-merging.enabled</code> is enabled.<br />Whether to allow merging data of multiple checkpoints into one physical file. If this option is set to false, only merge files within checkpoint boundaries. Otherwise, it is possible for the logical files of different checkpoints to share the same physical file.</td> + <td>Only relevant if <code class="highlighter-rouge">execution.checkpointing.file-merging.enabled</code> is enabled.<br />Whether to allow merging data of multiple checkpoints into one physical file. If this option is set to false, only merge files within checkpoint boundaries. Otherwise, it is possible for the logical files of different checkpoints to share the same physical file.</td> </tr> <tr> - <td><h5>state.checkpoints.file-merging.max-file-size</h5></td> + <td><h5>execution.checkpointing.file-merging.max-file-size</h5></td> <td style="word-wrap: break-word;">32 mb</td> <td>MemorySize</td> <td>Max size of a physical file for merged checkpoints.</td> </tr> <tr> - <td><h5>state.checkpoints.file-merging.pool-blocking</h5></td> + <td><h5>execution.checkpointing.file-merging.pool-blocking</h5></td> <td style="word-wrap: break-word;">false</td> <td>Boolean</td> <td>Whether to use Blocking or Non-Blocking pool for merging physical files. A Non-Blocking pool will always provide usable physical file without blocking. It may create many physical files if poll file frequently. When poll a small file from a Blocking pool, it may be blocked until the file is returned.</td> </tr> <tr> - <td><h5>state.checkpoints.file-merging.max-space-amplification</h5></td> + <td><h5>execution.checkpointing.file-merging.max-space-amplification</h5></td> <td style="word-wrap: break-word;">2.0</td> <td>Float</td> <td>Space amplification stands for the magnification of the occupied space compared to the amount of valid data. The more space amplification is, the more waste of space will be. This configs a space amplification above which a re-uploading for physical files will be triggered to reclaim space. Any value below 1f means disabling the space control.</td> diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html b/docs/layouts/shortcodes/generated/checkpointing_configuration.html index 9663e31b325..5fbf346b114 100644 --- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html +++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html @@ -15,16 +15,76 @@ <td>Only relevant if <code class="highlighter-rouge">execution.checkpointing.unaligned.enabled</code> is enabled.<br /><br />If timeout is 0, checkpoints will always start unaligned.<br /><br />If timeout has a positive value, checkpoints will start aligned. If during checkpointing, checkpoint start delay exceeds this timeout, alignment will timeout and checkpoint barrier will start working as unaligned checkpoint.</td> </tr> <tr> - <td><h5>execution.checkpointing.checkpoints-after-tasks-finish.enabled</h5></td> + <td><h5>execution.checkpointing.checkpoints-after-tasks-finish</h5></td> <td style="word-wrap: break-word;">true</td> <td>Boolean</td> <td>Feature toggle for enabling checkpointing even if some of tasks have finished. Before you enable it, please take a look at <a href="{{.Site.BaseURL}}{{.Site.LanguagePrefix}}/docs/dev/datastream/fault-tolerance/checkpointing/#checkpointing-with-parts-of-the-graph-finished-beta">the important considerations</a> </td> </tr> + <tr> + <td><h5>execution.checkpointing.cleaner.parallel-mode</h5></td> + <td style="word-wrap: break-word;">true</td> + <td>Boolean</td> + <td>Option whether to discard a checkpoint's states in parallel using the ExecutorService passed into the cleaner</td> + </tr> + <tr> + <td><h5>execution.checkpointing.create-subdir</h5></td> + <td style="word-wrap: break-word;">true</td> + <td>Boolean</td> + <td>Whether to create sub-directories named by job id under the '<code class="highlighter-rouge">execution.checkpointing.dir</code>' to store the data files and meta data of checkpoints. The default value is true to enable user could run several jobs with the same checkpoint directory at the same time. If this value is set to false, pay attention not to run several jobs with the same directory simultaneously. <br />WARNING: This is an advanced configuration. If set to false, [...] + </tr> + <tr> + <td><h5>execution.checkpointing.data-inline-threshold</h5></td> + <td style="word-wrap: break-word;">20 kb</td> + <td>MemorySize</td> + <td>The minimum size of state data files. All state chunks smaller than that are stored inline in the root checkpoint metadata file. The max memory threshold for this configuration is 1MB.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.dir</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The default directory used for storing the data files and meta data of checkpoints in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes(i.e. all TaskManagers and JobManagers). If the 'execution.checkpointing.storage' is set to 'jobmanager', only the meta data of checkpoints will be stored in this directory.</td> + </tr> <tr> <td><h5>execution.checkpointing.externalized-checkpoint-retention</h5></td> <td style="word-wrap: break-word;">NO_EXTERNALIZED_CHECKPOINTS</td> <td><p>Enum</p></td> - <td>Externalized checkpoints write their meta data out to persistent storage and are not automatically cleaned up when the owning job fails or is suspended (terminating with job status <code class="highlighter-rouge">JobStatus#FAILED</code> or <code class="highlighter-rouge">JobStatus#SUSPENDED</code>). In this case, you have to manually clean up the checkpoint state, both the meta data and actual program state.<br /><br />The mode defines how an externalized checkpoint shoul [...] + <td>Externalized checkpoints write their meta data out to persistent storage and are not automatically cleaned up when the owning job fails or is suspended (terminating with job status <code class="highlighter-rouge">JobStatus#FAILED</code> or <code class="highlighter-rouge">JobStatus#SUSPENDED</code>). In this case, you have to manually clean up the checkpoint state, both the meta data and actual program state.<br /><br />The mode defines how an externalized checkpoint shoul [...] + </tr> + <tr> + <td><h5>execution.checkpointing.file-merging.across-checkpoint-boundary</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Only relevant if <code class="highlighter-rouge">execution.checkpointing.file-merging.enabled</code> is enabled.<br />Whether to allow merging data of multiple checkpoints into one physical file. If this option is set to false, only merge files within checkpoint boundaries. Otherwise, it is possible for the logical files of different checkpoints to share the same physical file.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.file-merging.enabled</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Whether to enable merging multiple checkpoint files into one, which will greatly reduce the number of small checkpoint files. This is an experimental feature under evaluation, make sure you're aware of the possible effects of enabling it.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.file-merging.max-file-size</h5></td> + <td style="word-wrap: break-word;">32 mb</td> + <td>MemorySize</td> + <td>Max size of a physical file for merged checkpoints.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.file-merging.max-space-amplification</h5></td> + <td style="word-wrap: break-word;">2.0</td> + <td>Float</td> + <td>Space amplification stands for the magnification of the occupied space compared to the amount of valid data. The more space amplification is, the more waste of space will be. This configs a space amplification above which a re-uploading for physical files will be triggered to reclaim space. Any value below 1f means disabling the space control.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.file-merging.pool-blocking</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Whether to use Blocking or Non-Blocking pool for merging physical files. A Non-Blocking pool will always provide usable physical file without blocking. It may create many physical files if poll file frequently. When poll a small file from a Blocking pool, it may be blocked until the file is returned.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.incremental</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Option whether to create incremental checkpoints, if possible. For an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the complete checkpoint state. Once enabled, the state size shown in web UI or fetched from rest API only represents the delta checkpoint size instead of full checkpoint size. Some state backends may not support incremental checkpoints and ignore this option.</td> </tr> <tr> <td><h5>execution.checkpointing.interval</h5></td> @@ -38,6 +98,12 @@ <td>Duration</td> <td>If it is not null and any source reports isProcessingBacklog=true, it is the interval in which checkpoints are periodically scheduled.<br /><br />Checkpoint triggering may be delayed by the settings <code class="highlighter-rouge">execution.checkpointing.max-concurrent-checkpoints</code> and <code class="highlighter-rouge">execution.checkpointing.min-pause</code>.<br /><br />Note: if it is not null, the value must either be 0, which means the checkpoint is disabled during [...] </tr> + <tr> + <td><h5>execution.checkpointing.local-backup.dirs</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The config parameter defining the root directories for storing file-based state for local recovery. Local recovery currently only covers keyed state backends. If not configured it will default to <WORKING_DIR>/localState. The <WORKING_DIR> can be configured via <code class="highlighter-rouge">process.taskmanager.working-dir</code></td> + </tr> <tr> <td><h5>execution.checkpointing.local-backup.enabled</h5></td> <td style="word-wrap: break-word;">false</td> @@ -62,6 +128,24 @@ <td><p>Enum</p></td> <td>The checkpointing mode (exactly-once vs. at-least-once).<br /><br />Possible values:<ul><li>"EXACTLY_ONCE"</li><li>"AT_LEAST_ONCE"</li></ul></td> </tr> + <tr> + <td><h5>execution.checkpointing.num-retained</h5></td> + <td style="word-wrap: break-word;">1</td> + <td>Integer</td> + <td>The maximum number of completed checkpoints to retain.</td> + </tr> + <tr> + <td><h5>execution.checkpointing.savepoint-dir</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The default directory for savepoints. Used by the state backends that write savepoints to file systems (HashMapStateBackend, EmbeddedRocksDBStateBackend).</td> + </tr> + <tr> + <td><h5>execution.checkpointing.storage</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The checkpoint storage implementation to be used to checkpoint state.<br />The implementation can be specified either via their shortcut name, or via the class name of a <code class="highlighter-rouge">CheckpointStorageFactory</code>. If a factory is specified it is instantiated via its zero argument constructor and its <code class="highlighter-rouge">CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader)</code> method is called.<br />Recognized shortcut [...] + </tr> <tr> <td><h5>execution.checkpointing.timeout</h5></td> <td style="word-wrap: break-word;">10 min</td> @@ -99,94 +183,10 @@ <td>Defines the maximum number of subtasks that share the same channel state file. It can reduce the number of small files when enable unaligned checkpoint. Each subtask will create a new channel state file when this is configured to 1.</td> </tr> <tr> - <td><h5>state.backend.incremental</h5></td> - <td style="word-wrap: break-word;">false</td> - <td>Boolean</td> - <td>Option whether the state backend should create incremental checkpoints, if possible. For an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the complete checkpoint state. Once enabled, the state size shown in web UI or fetched from rest API only represents the delta checkpoint size instead of full checkpoint size. Some state backends may not support incremental checkpoints and ignore this option.</td> - </tr> - <tr> - <td><h5>state.checkpoint-storage</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>String</td> - <td>The checkpoint storage implementation to be used to checkpoint state.<br />The implementation can be specified either via their shortcut name, or via the class name of a <code class="highlighter-rouge">CheckpointStorageFactory</code>. If a factory is specified it is instantiated via its zero argument constructor and its <code class="highlighter-rouge">CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader)</code> method is called.<br />Recognized shortcut [...] - </tr> - <tr> - <td><h5>state.checkpoint.cleaner.parallel-mode</h5></td> - <td style="word-wrap: break-word;">true</td> - <td>Boolean</td> - <td>Option whether to discard a checkpoint's states in parallel using the ExecutorService passed into the cleaner</td> - </tr> - <tr> - <td><h5>state.checkpoints.create-subdir</h5></td> - <td style="word-wrap: break-word;">true</td> - <td>Boolean</td> - <td>Whether to create sub-directories named by job id under the '<code class="highlighter-rouge">state.checkpoints.dir</code>' to store the data files and meta data of checkpoints. The default value is true to enable user could run several jobs with the same checkpoint directory at the same time. If this value is set to false, pay attention not to run several jobs with the same directory simultaneously. <br />WARNING: This is an advanced configuration. If set to false, users [...] - </tr> - <tr> - <td><h5>state.checkpoints.dir</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>String</td> - <td>The default directory used for storing the data files and meta data of checkpoints in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes(i.e. all TaskManagers and JobManagers). If the 'state.checkpoint-storage' is set to 'jobmanager', only the meta data of checkpoints will be stored in this directory.</td> - </tr> - <tr> - <td><h5>state.checkpoints.file-merging.across-checkpoint-boundary</h5></td> - <td style="word-wrap: break-word;">false</td> - <td>Boolean</td> - <td>Only relevant if <code class="highlighter-rouge">state.checkpoints.file-merging.enabled</code> is enabled.<br />Whether to allow merging data of multiple checkpoints into one physical file. If this option is set to false, only merge files within checkpoint boundaries. Otherwise, it is possible for the logical files of different checkpoints to share the same physical file.</td> - </tr> - <tr> - <td><h5>state.checkpoints.file-merging.enabled</h5></td> - <td style="word-wrap: break-word;">false</td> - <td>Boolean</td> - <td>Whether to enable merging multiple checkpoint files into one, which will greatly reduce the number of small checkpoint files. This is an experimental feature under evaluation, make sure you're aware of the possible effects of enabling it.</td> - </tr> - <tr> - <td><h5>state.checkpoints.file-merging.max-file-size</h5></td> - <td style="word-wrap: break-word;">32 mb</td> - <td>MemorySize</td> - <td>Max size of a physical file for merged checkpoints.</td> - </tr> - <tr> - <td><h5>state.checkpoints.file-merging.max-space-amplification</h5></td> - <td style="word-wrap: break-word;">2.0</td> - <td>Float</td> - <td>Space amplification stands for the magnification of the occupied space compared to the amount of valid data. The more space amplification is, the more waste of space will be. This configs a space amplification above which a re-uploading for physical files will be triggered to reclaim space. Any value below 1f means disabling the space control.</td> - </tr> - <tr> - <td><h5>state.checkpoints.file-merging.pool-blocking</h5></td> - <td style="word-wrap: break-word;">false</td> - <td>Boolean</td> - <td>Whether to use Blocking or Non-Blocking pool for merging physical files. A Non-Blocking pool will always provide usable physical file without blocking. It may create many physical files if poll file frequently. When poll a small file from a Blocking pool, it may be blocked until the file is returned.</td> - </tr> - <tr> - <td><h5>state.checkpoints.num-retained</h5></td> - <td style="word-wrap: break-word;">1</td> - <td>Integer</td> - <td>The maximum number of completed checkpoints to retain.</td> - </tr> - <tr> - <td><h5>state.savepoints.dir</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>String</td> - <td>The default directory for savepoints. Used by the state backends that write savepoints to file systems (HashMapStateBackend, EmbeddedRocksDBStateBackend).</td> - </tr> - <tr> - <td><h5>state.storage.fs.memory-threshold</h5></td> - <td style="word-wrap: break-word;">20 kb</td> - <td>MemorySize</td> - <td>The minimum size of state data files. All state chunks smaller than that are stored inline in the root checkpoint metadata file. The max memory threshold for this configuration is 1MB.</td> - </tr> - <tr> - <td><h5>state.storage.fs.write-buffer-size</h5></td> + <td><h5>execution.checkpointing.write-buffer-size</h5></td> <td style="word-wrap: break-word;">4096</td> <td>Integer</td> - <td>The default size of the write buffer for the checkpoint streams that write to file systems. The actual write buffer size is determined to be the maximum of the value of this option and option 'state.storage.fs.memory-threshold'.</td> - </tr> - <tr> - <td><h5>taskmanager.state.local.root-dirs</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>String</td> - <td>The config parameter defining the root directories for storing file-based state for local recovery. Local recovery currently only covers keyed state backends. If not configured it will default to <WORKING_DIR>/localState. The <WORKING_DIR> can be configured via <code class="highlighter-rouge">process.taskmanager.working-dir</code></td> + <td>The default size of the write buffer for the checkpoint streams that write to file systems. The actual write buffer size is determined to be the maximum of the value of this option and option 'execution.checkpointing.data-inline-threshold'.</td> </tr> </tbody> </table> diff --git a/docs/layouts/shortcodes/generated/common_state_backends_section.html b/docs/layouts/shortcodes/generated/common_state_backends_section.html index ab664f51bfb..7fe3a46207a 100644 --- a/docs/layouts/shortcodes/generated/common_state_backends_section.html +++ b/docs/layouts/shortcodes/generated/common_state_backends_section.html @@ -15,52 +15,52 @@ <td>The state backend to be used to store state.<br />The implementation can be specified either via their shortcut name, or via the class name of a <code class="highlighter-rouge">StateBackendFactory</code>. If a factory is specified it is instantiated via its zero argument constructor and its <code class="highlighter-rouge">StateBackendFactory#createFromConfig(ReadableConfig, ClassLoader)</code> method is called.<br />Recognized shortcut names are 'hashmap' and 'rocksdb'.</td> </tr> <tr> - <td><h5>state.checkpoint-storage</h5></td> + <td><h5>execution.checkpointing.dir</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>String</td> - <td>The checkpoint storage implementation to be used to checkpoint state.<br />The implementation can be specified either via their shortcut name, or via the class name of a <code class="highlighter-rouge">CheckpointStorageFactory</code>. If a factory is specified it is instantiated via its zero argument constructor and its <code class="highlighter-rouge">CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader)</code> method is called.<br />Recognized shortcut [...] + <td>The default directory used for storing the data files and meta data of checkpoints in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes(i.e. all TaskManagers and JobManagers). If the 'execution.checkpointing.storage' is set to 'jobmanager', only the meta data of checkpoints will be stored in this directory.</td> </tr> <tr> - <td><h5>state.checkpoints.dir</h5></td> + <td><h5>execution.checkpointing.storage</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>String</td> - <td>The default directory used for storing the data files and meta data of checkpoints in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes(i.e. all TaskManagers and JobManagers). If the 'state.checkpoint-storage' is set to 'jobmanager', only the meta data of checkpoints will be stored in this directory.</td> + <td>The checkpoint storage implementation to be used to checkpoint state.<br />The implementation can be specified either via their shortcut name, or via the class name of a <code class="highlighter-rouge">CheckpointStorageFactory</code>. If a factory is specified it is instantiated via its zero argument constructor and its <code class="highlighter-rouge">CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader)</code> method is called.<br />Recognized shortcut [...] </tr> <tr> - <td><h5>state.savepoints.dir</h5></td> + <td><h5>execution.checkpointing.savepoint-dir</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>String</td> <td>The default directory for savepoints. Used by the state backends that write savepoints to file systems (HashMapStateBackend, EmbeddedRocksDBStateBackend).</td> </tr> <tr> - <td><h5>execution.state-recovery.from-local</h5></td> - <td style="word-wrap: break-word;">false</td> + <td><h5>execution.checkpointing.cleaner.parallel-mode</h5></td> + <td style="word-wrap: break-word;">true</td> <td>Boolean</td> - <td>This option configures local recovery for the state backend, which indicates whether to recovery from local snapshot.By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend)."</td> + <td>Option whether to discard a checkpoint's states in parallel using the ExecutorService passed into the cleaner</td> </tr> <tr> - <td><h5>state.backend.incremental</h5></td> + <td><h5>execution.checkpointing.incremental</h5></td> <td style="word-wrap: break-word;">false</td> <td>Boolean</td> - <td>Option whether the state backend should create incremental checkpoints, if possible. For an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the complete checkpoint state. Once enabled, the state size shown in web UI or fetched from rest API only represents the delta checkpoint size instead of full checkpoint size. Some state backends may not support incremental checkpoints and ignore this option.</td> + <td>Option whether to create incremental checkpoints, if possible. For an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the complete checkpoint state. Once enabled, the state size shown in web UI or fetched from rest API only represents the delta checkpoint size instead of full checkpoint size. Some state backends may not support incremental checkpoints and ignore this option.</td> </tr> <tr> - <td><h5>state.checkpoint.cleaner.parallel-mode</h5></td> - <td style="word-wrap: break-word;">true</td> - <td>Boolean</td> - <td>Option whether to discard a checkpoint's states in parallel using the ExecutorService passed into the cleaner</td> + <td><h5>execution.checkpointing.local-backup.dirs</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The config parameter defining the root directories for storing file-based state for local recovery. Local recovery currently only covers keyed state backends. If not configured it will default to <WORKING_DIR>/localState. The <WORKING_DIR> can be configured via <code class="highlighter-rouge">process.taskmanager.working-dir</code></td> </tr> <tr> - <td><h5>state.checkpoints.num-retained</h5></td> + <td><h5>execution.checkpointing.num-retained</h5></td> <td style="word-wrap: break-word;">1</td> <td>Integer</td> <td>The maximum number of completed checkpoints to retain.</td> </tr> <tr> - <td><h5>taskmanager.state.local.root-dirs</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>String</td> - <td>The config parameter defining the root directories for storing file-based state for local recovery. Local recovery currently only covers keyed state backends. If not configured it will default to <WORKING_DIR>/localState. The <WORKING_DIR> can be configured via <code class="highlighter-rouge">process.taskmanager.working-dir</code></td> + <td><h5>execution.state-recovery.from-local</h5></td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>This option configures local recovery for the state backend, which indicates whether to recovery from local snapshot.By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend)."</td> </tr> </tbody> </table> diff --git a/docs/layouts/shortcodes/generated/expert_state_backends_section.html b/docs/layouts/shortcodes/generated/expert_state_backends_section.html index d496aca74b5..582a61f89f3 100644 --- a/docs/layouts/shortcodes/generated/expert_state_backends_section.html +++ b/docs/layouts/shortcodes/generated/expert_state_backends_section.html @@ -9,22 +9,22 @@ </thead> <tbody> <tr> - <td><h5>state.checkpoints.create-subdir</h5></td> + <td><h5>execution.checkpointing.create-subdir</h5></td> <td style="word-wrap: break-word;">true</td> <td>Boolean</td> - <td>Whether to create sub-directories named by job id under the '<code class="highlighter-rouge">state.checkpoints.dir</code>' to store the data files and meta data of checkpoints. The default value is true to enable user could run several jobs with the same checkpoint directory at the same time. If this value is set to false, pay attention not to run several jobs with the same directory simultaneously. <br />WARNING: This is an advanced configuration. If set to false, users [...] + <td>Whether to create sub-directories named by job id under the '<code class="highlighter-rouge">execution.checkpointing.dir</code>' to store the data files and meta data of checkpoints. The default value is true to enable user could run several jobs with the same checkpoint directory at the same time. If this value is set to false, pay attention not to run several jobs with the same directory simultaneously. <br />WARNING: This is an advanced configuration. If set to false, [...] </tr> <tr> - <td><h5>state.storage.fs.memory-threshold</h5></td> + <td><h5>execution.checkpointing.data-inline-threshold</h5></td> <td style="word-wrap: break-word;">20 kb</td> <td>MemorySize</td> <td>The minimum size of state data files. All state chunks smaller than that are stored inline in the root checkpoint metadata file. The max memory threshold for this configuration is 1MB.</td> </tr> <tr> - <td><h5>state.storage.fs.write-buffer-size</h5></td> + <td><h5>execution.checkpointing.write-buffer-size</h5></td> <td style="word-wrap: break-word;">4096</td> <td>Integer</td> - <td>The default size of the write buffer for the checkpoint streams that write to file systems. The actual write buffer size is determined to be the maximum of the value of this option and option 'state.storage.fs.memory-threshold'.</td> + <td>The default size of the write buffer for the checkpoint streams that write to file systems. The actual write buffer size is determined to be the maximum of the value of this option and option 'execution.checkpointing.data-inline-threshold'.</td> </tr> </tbody> </table> diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java index ec7b82ecfd8..1311c554f4a 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java @@ -99,9 +99,10 @@ public class CheckpointingOptions { */ @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS, position = 2) public static final ConfigOption<String> CHECKPOINT_STORAGE = - ConfigOptions.key("state.checkpoint-storage") + ConfigOptions.key("execution.checkpointing.storage") .stringType() .noDefaultValue() + .withDeprecatedKeys("state.checkpoint-storage") .withDescription( Description.builder() .text( @@ -121,21 +122,22 @@ public class CheckpointingOptions { "Recognized shortcut names are 'jobmanager' and 'filesystem'.") .linebreak() .text( - "'state.checkpoint-storage' and 'state.checkpoints.dir' are usually combined to configure the checkpoint location." + "'execution.checkpointing.storage' and 'execution.checkpointing.dir' are usually combined to configure the checkpoint location." + " By default, the checkpoint meta data and actual program state will be stored in the JobManager's memory directly." - + " When 'state.checkpoint-storage' is set to 'jobmanager', if 'state.checkpoints.dir' is configured," - + " the meta data of checkpoints will be persisted to the path specified by 'state.checkpoints.dir'." + + " When 'execution.checkpointing.storage' is set to 'jobmanager', if 'execution.checkpointing.dir' is configured," + + " the meta data of checkpoints will be persisted to the path specified by 'execution.checkpointing.dir'." + " Otherwise, the meta data will be stored in the JobManager's memory." - + " When 'state.checkpoint-storage' is set to 'filesystem', a valid path must be configured to 'state.checkpoints.dir'," + + " When 'execution.checkpointing.storage' is set to 'filesystem', a valid path must be configured to 'execution.checkpointing.dir'," + " and the checkpoint meta data and actual program state will both be persisted to the path.") .build()); /** The maximum number of completed checkpoints to retain. */ @Documentation.Section(Documentation.Sections.COMMON_STATE_BACKENDS) public static final ConfigOption<Integer> MAX_RETAINED_CHECKPOINTS = - ConfigOptions.key("state.checkpoints.num-retained") + ConfigOptions.key("execution.checkpointing.num-retained") .intType() .defaultValue(1) + .withDeprecatedKeys("state.checkpoints.num-retained") .withDescription("The maximum number of completed checkpoints to retain."); /* Option whether to clean individual checkpoint's operatorstates in parallel. If enabled, @@ -144,9 +146,10 @@ public class CheckpointingOptions { */ @Documentation.Section(Documentation.Sections.COMMON_STATE_BACKENDS) public static final ConfigOption<Boolean> CLEANER_PARALLEL_MODE = - ConfigOptions.key("state.checkpoint.cleaner.parallel-mode") + ConfigOptions.key("execution.checkpointing.cleaner.parallel-mode") .booleanType() .defaultValue(true) + .withDeprecatedKeys("state.checkpoint.cleaner.parallel-mode") .withDescription( "Option whether to discard a checkpoint's states in parallel using" + " the ExecutorService passed into the cleaner"); @@ -160,9 +163,9 @@ public class CheckpointingOptions { .withDescription("Deprecated option. All state snapshots are asynchronous."); /** - * Option whether the state backend should create incremental checkpoints, if possible. For an - * incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the - * complete checkpoint state. + * Option whether to create incremental checkpoints, if possible. For an incremental checkpoint, + * only a diff from the previous checkpoint is stored, rather than the complete checkpoint + * state. * * <p>Once enabled, the state size shown in web UI or fetched from rest API only represents the * delta checkpoint size instead of full checkpoint size. @@ -171,11 +174,12 @@ public class CheckpointingOptions { */ @Documentation.Section(Documentation.Sections.COMMON_STATE_BACKENDS) public static final ConfigOption<Boolean> INCREMENTAL_CHECKPOINTS = - ConfigOptions.key("state.backend.incremental") + ConfigOptions.key("execution.checkpointing.incremental") .booleanType() .defaultValue(false) + .withDeprecatedKeys("state.backend.incremental") .withDescription( - "Option whether the state backend should create incremental checkpoints, if possible. For" + "Option whether to create incremental checkpoints, if possible. For" + " an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the" + " complete checkpoint state. Once enabled, the state size shown in web UI or fetched from rest API" + " only represents the delta checkpoint size instead of full checkpoint size." @@ -212,9 +216,10 @@ public class CheckpointingOptions { */ @Documentation.Section(Documentation.Sections.COMMON_STATE_BACKENDS) public static final ConfigOption<String> LOCAL_RECOVERY_TASK_MANAGER_STATE_ROOT_DIRS = - ConfigOptions.key("taskmanager.state.local.root-dirs") + ConfigOptions.key("execution.checkpointing.local-backup.dirs") .stringType() .noDefaultValue() + .withDeprecatedKeys("taskmanager.state.local.root-dirs") .withDescription( Description.builder() .text( @@ -238,10 +243,10 @@ public class CheckpointingOptions { */ @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS, position = 3) public static final ConfigOption<String> SAVEPOINT_DIRECTORY = - ConfigOptions.key("state.savepoints.dir") + ConfigOptions.key("execution.checkpointing.savepoint-dir") .stringType() .noDefaultValue() - .withDeprecatedKeys("savepoints.state.backend.fs.dir") + .withDeprecatedKeys("state.savepoints.dir", "savepoints.state.backend.fs.dir") .withDescription( "The default directory for savepoints. Used by the state backends that write savepoints to" + " file systems (HashMapStateBackend, EmbeddedRocksDBStateBackend)."); @@ -254,10 +259,10 @@ public class CheckpointingOptions { */ @Documentation.Section(value = Documentation.Sections.COMMON_STATE_BACKENDS, position = 2) public static final ConfigOption<String> CHECKPOINTS_DIRECTORY = - ConfigOptions.key("state.checkpoints.dir") + ConfigOptions.key("execution.checkpointing.dir") .stringType() .noDefaultValue() - .withDeprecatedKeys("state.backend.fs.checkpointdir") + .withDeprecatedKeys("state.checkpoints.dir", "state.backend.fs.checkpointdir") .withDescription( "The default directory used for storing the data files and meta data of checkpoints " + "in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes" @@ -273,9 +278,10 @@ public class CheckpointingOptions { */ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS) public static final ConfigOption<Boolean> CREATE_CHECKPOINT_SUB_DIR = - ConfigOptions.key("state.checkpoints.create-subdir") + ConfigOptions.key("execution.checkpointing.create-subdir") .booleanType() .defaultValue(true) + .withDeprecatedKeys("state.checkpoints.create-subdir") .withDescription( Description.builder() .text( @@ -297,20 +303,22 @@ public class CheckpointingOptions { */ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS) public static final ConfigOption<MemorySize> FS_SMALL_FILE_THRESHOLD = - ConfigOptions.key("state.storage.fs.memory-threshold") + ConfigOptions.key("execution.checkpointing.data-inline-threshold") .memoryType() .defaultValue(MemorySize.parse("20kb")) .withDescription( "The minimum size of state data files. All state chunks smaller than that are stored" + " inline in the root checkpoint metadata file. The max memory threshold for this configuration is 1MB.") - .withDeprecatedKeys("state.backend.fs.memory-threshold"); + .withDeprecatedKeys( + "state.storage.fs.memory-threshold", + "state.backend.fs.memory-threshold"); /** * The default size of the write buffer for the checkpoint streams that write to file systems. */ @Documentation.Section(Documentation.Sections.EXPERT_STATE_BACKENDS) public static final ConfigOption<Integer> FS_WRITE_BUFFER_SIZE = - ConfigOptions.key("state.storage.fs.write-buffer-size") + ConfigOptions.key("execution.checkpointing.write-buffer-size") .intType() .defaultValue(4 * 1024) .withDescription( @@ -318,7 +326,9 @@ public class CheckpointingOptions { "The default size of the write buffer for the checkpoint streams that write to file systems. " + "The actual write buffer size is determined to be the maximum of the value of this option and option '%s'.", FS_SMALL_FILE_THRESHOLD.key())) - .withDeprecatedKeys("state.backend.fs.write-buffer-size"); + .withDeprecatedKeys( + "state.storage.fs.write-buffer-size", + "state.backend.fs.write-buffer-size"); /** * This option configures local backup for the state backend, which indicates whether to make @@ -355,7 +365,7 @@ public class CheckpointingOptions { @Experimental @Documentation.Section(value = Documentation.Sections.CHECKPOINT_FILE_MERGING, position = 1) public static final ConfigOption<Boolean> FILE_MERGING_ENABLED = - ConfigOptions.key("state.checkpoints.file-merging.enabled") + ConfigOptions.key("execution.checkpointing.file-merging.enabled") .booleanType() .defaultValue(false) .withDescription( @@ -371,7 +381,7 @@ public class CheckpointingOptions { @Experimental @Documentation.Section(value = Documentation.Sections.CHECKPOINT_FILE_MERGING, position = 2) public static final ConfigOption<Boolean> FILE_MERGING_ACROSS_BOUNDARY = - ConfigOptions.key("state.checkpoints.file-merging.across-checkpoint-boundary") + ConfigOptions.key("execution.checkpointing.file-merging.across-checkpoint-boundary") .booleanType() .defaultValue(false) .withDescription( @@ -392,7 +402,7 @@ public class CheckpointingOptions { @Experimental @Documentation.Section(value = Documentation.Sections.CHECKPOINT_FILE_MERGING, position = 3) public static final ConfigOption<MemorySize> FILE_MERGING_MAX_FILE_SIZE = - ConfigOptions.key("state.checkpoints.file-merging.max-file-size") + ConfigOptions.key("execution.checkpointing.file-merging.max-file-size") .memoryType() .defaultValue(MemorySize.parse("32MB")) .withDescription("Max size of a physical file for merged checkpoints."); @@ -406,7 +416,7 @@ public class CheckpointingOptions { @Experimental @Documentation.Section(value = Documentation.Sections.CHECKPOINT_FILE_MERGING, position = 4) public static final ConfigOption<Boolean> FILE_MERGING_POOL_BLOCKING = - ConfigOptions.key("state.checkpoints.file-merging.pool-blocking") + ConfigOptions.key("execution.checkpointing.file-merging.pool-blocking") .booleanType() .defaultValue(false) .withDescription( @@ -422,7 +432,7 @@ public class CheckpointingOptions { */ @Experimental @Documentation.ExcludeFromDocumentation public static final ConfigOption<Integer> FILE_MERGING_MAX_SUBTASKS_PER_FILE = - ConfigOptions.key("state.checkpoints.file-merging.max-subtasks-per-file") + ConfigOptions.key("execution.checkpointing.file-merging.max-subtasks-per-file") .intType() .defaultValue(4) .withDescription( @@ -438,7 +448,7 @@ public class CheckpointingOptions { @Experimental @Documentation.Section(value = Documentation.Sections.CHECKPOINT_FILE_MERGING, position = 6) public static final ConfigOption<Float> FILE_MERGING_MAX_SPACE_AMPLIFICATION = - ConfigOptions.key("state.checkpoints.file-merging.max-space-amplification") + ConfigOptions.key("execution.checkpointing.file-merging.max-space-amplification") .floatType() .defaultValue(2f) .withDescription( @@ -641,9 +651,11 @@ public class CheckpointingOptions { "Allows unaligned checkpoints to skip timers that are currently being fired."); public static final ConfigOption<Boolean> ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH = - ConfigOptions.key("execution.checkpointing.checkpoints-after-tasks-finish.enabled") + ConfigOptions.key("execution.checkpointing.checkpoints-after-tasks-finish") .booleanType() .defaultValue(true) + .withDeprecatedKeys( + "execution.checkpointing.checkpoints-after-tasks-finish.enabled") .withDescription( Description.builder() .text( diff --git a/flink-end-to-end-tests/test-scripts/test_file_sink.sh b/flink-end-to-end-tests/test-scripts/test_file_sink.sh index 204c6442b7e..0acc9425bcb 100755 --- a/flink-end-to-end-tests/test-scripts/test_file_sink.sh +++ b/flink-end-to-end-tests/test-scripts/test_file_sink.sh @@ -69,7 +69,7 @@ elif [ "${OUT_TYPE}" == "s3" ]; then S3_DATA_PREFIX="${RANDOM_PREFIX}" S3_CHECKPOINT_PREFIX="${RANDOM_PREFIX}-chk" JOB_OUTPUT_PATH="s3://$IT_CASE_S3_BUCKET/${S3_DATA_PREFIX}" - set_config_key "state.checkpoints.dir" "s3://$IT_CASE_S3_BUCKET/${S3_CHECKPOINT_PREFIX}" + set_config_key "execution.checkpointing.dir" "s3://$IT_CASE_S3_BUCKET/${S3_CHECKPOINT_PREFIX}" # overwrites implementation for local runs function get_complete_result { diff --git a/flink-python/pyflink/datastream/state_backend.py b/flink-python/pyflink/datastream/state_backend.py index 7b4c69917bb..703a3b9d6b6 100644 --- a/flink-python/pyflink/datastream/state_backend.py +++ b/flink-python/pyflink/datastream/state_backend.py @@ -388,7 +388,7 @@ class MemoryStateBackend(StateBackend): ``` state.backend.type: hashmap - state.checkpoint-storage: jobmanager + execution.checkpointing.storage: jobmanager ``` This state backend holds the working state in the memory (JVM heap) of the TaskManagers. diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java index f2beb9033a2..78917e0623b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointSubsumeHelper.java @@ -26,8 +26,8 @@ import java.util.Optional; /** * Encapsulates the logic to subsume older checkpoints by {@link CompletedCheckpointStore checkpoint - * stores}. In general, checkpoints should be subsumed whenever state.checkpoints.num-retained is - * exceeded. + * stores}. In general, checkpoints should be subsumed whenever execution.checkpointing.num-retained + * is exceeded. * * <p>Additional considerations: * diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java index df4809714ef..8d054e1de13 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java @@ -48,7 +48,8 @@ public class CheckpointStorageLoader { /** * Loads the checkpoint storage from the configuration, from the parameter - * 'state.checkpoint-storage', as defined in {@link CheckpointingOptions#CHECKPOINT_STORAGE}. + * 'execution.checkpointing.storage', as defined in {@link + * CheckpointingOptions#CHECKPOINT_STORAGE}. * * <p>The implementation can be specified either via their shortcut name, or via the class name * of a {@link CheckpointStorageFactory}. If a CheckpointStorageFactory class name is specified, diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java index 6345c2b4602..6ffbfd98230 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemoryStateBackend.java @@ -64,7 +64,7 @@ import static org.apache.flink.util.Preconditions.checkArgument; * * <pre>{@code * state.backend.type: hashmap - * state.checkpoint-storage: jobmanager + * execution.checkpointing.storage: jobmanager * }</pre> * * <p>This state backend holds the working state in the memory (JVM heap) of the TaskManagers. The diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java index 8c033d833a2..3eece8c95d8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/SavepointHandlersTest.java @@ -171,7 +171,7 @@ class SavepointHandlersTest { } catch (RestHandlerException rhe) { assertThat(rhe.getMessage()) .isEqualTo( - "Config key [state.savepoints.dir] is not set. " + "Config key [execution.checkpointing.savepoint-dir] is not set. " + "Property [target-directory] must be provided."); assertThat(rhe.getHttpResponseStatus()).isEqualTo(HttpResponseStatus.BAD_REQUEST); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/StopWithSavepointHandlersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/StopWithSavepointHandlersTest.java index d4649d68ed9..55891bd5c66 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/StopWithSavepointHandlersTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/savepoints/StopWithSavepointHandlersTest.java @@ -172,7 +172,7 @@ class StopWithSavepointHandlersTest { } catch (RestHandlerException rhe) { assertThat(rhe.getMessage()) .isEqualTo( - "Config key [state.savepoints.dir] is not set. " + "Config key [execution.checkpointing.savepoint-dir] is not set. " + "Property [targetDirectory] must be provided."); assertThat(rhe.getHttpResponseStatus()).isEqualTo(HttpResponseStatus.BAD_REQUEST); } diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java index a5b561ec009..961a949ec83 100644 --- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java +++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/MaterializedTableStatementITCase.java @@ -498,7 +498,8 @@ public class MaterializedTableStatementITCase extends AbstractMaterializedTableS // set up savepoint dir String savepointDir = temporaryPath.toString(); String alterJobSavepointDDL = - String.format("SET 'state.savepoints.dir' = 'file://%s'", savepointDir); + String.format( + "SET 'execution.checkpointing.savepoint-dir' = 'file://%s'", savepointDir); OperationHandle alterMaterializedTableSavepointHandle = service.executeStatement( sessionHandle, alterJobSavepointDDL, -1, new Configuration());
