You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/03/14 09:30:42 UTC
(flink) 01/03: [FLINK-34484][state]Move LOCAL_RECOVERY config from CheckpointingOptions to StateRecoveryOption
This is an automated email from the ASF dual-hosted git repository.
leiyanfei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1d1dca6d6b1a80c4fcd3337489a99c0d834692ea
Author: Jinzhong Li <li...@gmail.com>
AuthorDate: Tue Feb 27 18:08:03 2024 +0800
[FLINK-34484][state]Move LOCAL_RECOVERY config from CheckpointingOptions to StateRecoveryOption
---
.../generated/checkpointing_configuration.html | 6 ------
.../generated/common_state_backends_section.html | 6 ------
.../generated/state_recovery_configuration.html | 6 ++++++
.../flink/configuration/CheckpointingOptions.java | 4 ++++
.../flink/configuration/StateRecoveryOptions.java | 17 +++++++++++++++++
.../flink/streaming/tests/AllroundMiniClusterTest.java | 4 ++--
.../DefaultSlotPoolServiceSchedulerFactory.java | 3 ++-
.../scheduler/ExecutionSlotAllocationContext.java | 3 +--
.../taskexecutor/TaskManagerServicesConfiguration.java | 3 ++-
.../flink/runtime/util/SlotSelectionStrategyUtils.java | 4 ++--
.../DefaultSlotPoolServiceSchedulerFactoryTest.java | 4 ++--
.../state/TaskExecutorLocalStateStoresManagerTest.java | 3 ++-
.../runtime/taskexecutor/TaskExecutorRecoveryTest.java | 4 ++--
.../runtime/util/SlotSelectionStrategyUtilsTest.java | 6 +++---
.../flink/test/checkpointing/AutoRescalingITCase.java | 3 ++-
.../checkpointing/ChangelogLocalRecoveryITCase.java | 2 +-
.../flink/test/checkpointing/LocalRecoveryITCase.java | 4 ++--
.../checkpointing/NotifyCheckpointAbortedITCase.java | 4 ++--
.../checkpointing/ResumeCheckpointManuallyITCase.java | 3 ++-
.../apache/flink/test/recovery/LocalRecoveryITCase.java | 4 ++--
.../runtime/DefaultSchedulerLocalRecoveryITCase.java | 4 ++--
.../org/apache/flink/test/runtime/SchedulingITCase.java | 6 +++---
.../test/state/ChangelogRecoveryCachingITCase.java | 2 +-
.../flink/test/state/ChangelogRescalingITCase.java | 2 +-
24 files changed, 63 insertions(+), 44 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
index 2c40c7e1804..0cd3dd53755 100644
--- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
@@ -14,12 +14,6 @@
<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.backend.local-recovery</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>This option configures local recovery for this state backend. By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend).</td>
- </tr>
<tr>
<td><h5>state.checkpoint-storage</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/layouts/shortcodes/generated/common_state_backends_section.html b/docs/layouts/shortcodes/generated/common_state_backends_section.html
index 7d35697849c..a16650bf412 100644
--- a/docs/layouts/shortcodes/generated/common_state_backends_section.html
+++ b/docs/layouts/shortcodes/generated/common_state_backends_section.html
@@ -38,12 +38,6 @@
<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.backend.local-recovery</h5></td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>This option configures local recovery for this state backend. By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend).</td>
- </tr>
<tr>
<td><h5>state.checkpoint.cleaner.parallel-mode</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git a/docs/layouts/shortcodes/generated/state_recovery_configuration.html b/docs/layouts/shortcodes/generated/state_recovery_configuration.html
index 5c255b2f820..2a1d914e2fd 100644
--- a/docs/layouts/shortcodes/generated/state_recovery_configuration.html
+++ b/docs/layouts/shortcodes/generated/state_recovery_configuration.html
@@ -14,6 +14,12 @@
<td><p>Enum</p></td>
<td>Describes the mode how Flink should restore from the given savepoint or retained checkpoint.<br /><br />Possible values:<ul><li>"CLAIM": Flink will take ownership of the given snapshot. It will clean the snapshot once it is subsumed by newer ones.</li><li>"NO_CLAIM": Flink will not claim ownership of the snapshot files. However it will make sure it does not depend on any artefacts from the restored snapshot. In order to do that, Flink will take the first checkpoint as a f [...]
</tr>
+ <tr>
+ <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. By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend)."</td>
+ </tr>
<tr>
<td><h5>execution.state-recovery.ignore-unclaimed-state</h5></td>
<td style="word-wrap: break-word;">false</td>
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 8cbec13c2cb..a5a00e813a0 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
@@ -178,8 +178,12 @@ public class CheckpointingOptions {
*
* <p>Local recovery currently only covers keyed state backends (including both the
* EmbeddedRocksDBStateBackend and the HashMapStateBackend).
+ *
+ * @Deprecated use {@link StateRecoveryOptions#LOCAL_RECOVERY} instead.
*/
@Documentation.Section(Documentation.Sections.COMMON_STATE_BACKENDS)
+ @Documentation.ExcludeFromDocumentation("Hidden for deprecated")
+ @Deprecated
public static final ConfigOption<Boolean> LOCAL_RECOVERY =
ConfigOptions.key("state.backend.local-recovery")
.booleanType()
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/StateRecoveryOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/StateRecoveryOptions.java
index a2bbc2e354f..cfd585b89f6 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/StateRecoveryOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/StateRecoveryOptions.java
@@ -97,4 +97,21 @@ public class StateRecoveryOptions {
+ "the specific checkpoint without in-flight data.")
.linebreak()
.build());
+
+ /**
+ * This option configures local recovery for the state backend. By default, local recovery is
+ * deactivated.
+ *
+ * <p>Local recovery currently only covers keyed state backends (including both the
+ * EmbeddedRocksDBStateBackend and the HashMapStateBackend).
+ *
+ */
+ public static final ConfigOption<Boolean> LOCAL_RECOVERY =
+ ConfigOptions.key("execution.state-recovery.from-local")
+ .booleanType()
+ .defaultValue(false)
+ .withDeprecatedKeys(CheckpointingOptions.LOCAL_RECOVERY.key())
+ .withDescription(" This option configures local recovery for the state backend. "
+ + "By default, local recovery is deactivated. Local recovery currently only "
+ + "covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend).\"");
}
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/test/java/org/apache/flink/streaming/tests/AllroundMiniClusterTest.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/test/java/org/apache/flink/streaming/tests/AllroundMiniClusterTest.java
index 8778ab2a2fe..eabe48b0639 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/test/java/org/apache/flink/streaming/tests/AllroundMiniClusterTest.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/test/java/org/apache/flink/streaming/tests/AllroundMiniClusterTest.java
@@ -18,8 +18,8 @@
package org.apache.flink.streaming.tests;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.util.TestLogger;
@@ -57,7 +57,7 @@ public class AllroundMiniClusterTest extends TestLogger {
private static Configuration createConfiguration() {
Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
configuration.setString(EXECUTION_FAILOVER_STRATEGY.key(), "region");
return configuration;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
index 4514f890510..8e85563fdf9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactory.java
@@ -26,6 +26,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RpcOptions;
import org.apache.flink.configuration.SchedulerExecutionMode;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.core.failure.FailureEnricher;
import org.apache.flink.runtime.blob.BlobWriter;
import org.apache.flink.runtime.blocklist.BlocklistOperations;
@@ -265,7 +266,7 @@ public final class DefaultSlotPoolServiceSchedulerFactory
static RequestSlotMatchingStrategy getRequestSlotMatchingStrategy(
Configuration configuration, JobType jobType) {
final boolean isLocalRecoveryEnabled =
- configuration.get(CheckpointingOptions.LOCAL_RECOVERY);
+ configuration.get(StateRecoveryOptions.LOCAL_RECOVERY);
if (isLocalRecoveryEnabled) {
if (jobType == JobType.STREAMING) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionSlotAllocationContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionSlotAllocationContext.java
index 44f8956b4a5..41ba23cf5e2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionSlotAllocationContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionSlotAllocationContext.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.scheduler;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
@@ -75,7 +74,7 @@ interface ExecutionSlotAllocationContext extends InputsLocationsRetriever, State
* Returns all reserved allocations. These allocations/slots were used to run certain vertices
* and reserving them can prevent other vertices to take these slots and thus help vertices to
* be deployed into their previous slots again after failover. It is needed if {@link
- * CheckpointingOptions#LOCAL_RECOVERY} is enabled.
+ * org.apache.flink.configuration.StateRecoveryOptions#LOCAL_RECOVERY} is enabled.
*
* @return all reserved allocations
*/
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
index fbd5a456747..a6577c26b93 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfiguration.java
@@ -26,6 +26,7 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.RpcOptions;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.TaskManagerOptionsInternal;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -283,7 +284,7 @@ public class TaskManagerServicesConfiguration {
localStateDirs = Reference.owned(createdLocalStateDirs);
}
- boolean localRecoveryMode = configuration.get(CheckpointingOptions.LOCAL_RECOVERY);
+ boolean localRecoveryMode = configuration.get(StateRecoveryOptions.LOCAL_RECOVERY);
final QueryableStateConfiguration queryableStateConfig =
QueryableStateConfiguration.fromConfiguration(configuration);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtils.java
index c1c03439ae5..2bd433a0cfa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtils.java
@@ -19,8 +19,8 @@
package org.apache.flink.runtime.util;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobmaster.slotpool.LocationPreferenceSlotSelectionStrategy;
@@ -50,7 +50,7 @@ public class SlotSelectionStrategyUtils {
: LocationPreferenceSlotSelectionStrategy.createDefault();
final boolean isLocalRecoveryEnabled =
- configuration.get(CheckpointingOptions.LOCAL_RECOVERY);
+ configuration.get(StateRecoveryOptions.LOCAL_RECOVERY);
if (isLocalRecoveryEnabled) {
if (jobType == JobType.STREAMING) {
return PreviousAllocationSlotSelectionStrategy.create(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
index 5f798bf9ada..bdc42e2d5cb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
@@ -18,10 +18,10 @@
package org.apache.flink.runtime.jobmaster;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.SchedulerExecutionMode;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobmaster.slotpool.PreferredAllocationRequestSlotMatchingStrategy;
import org.apache.flink.runtime.jobmaster.slotpool.RequestSlotMatchingStrategy;
@@ -132,7 +132,7 @@ public class DefaultSlotPoolServiceSchedulerFactoryTest {
public void testGetRequestSlotMatchingStrategy(
boolean isLocalRecoveryEnabled, JobType jobType, RequestSlotMatchingStrategy expected) {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, isLocalRecoveryEnabled);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, isLocalRecoveryEnabled);
assertThat(
DefaultSlotPoolServiceSchedulerFactory.getRequestSlotMatchingStrategy(
configuration, jobType))
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
index c50dca11683..fe5bdb739f3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/TaskExecutorLocalStateStoresManagerTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.state;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.blob.VoidPermanentBlobService;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -81,7 +82,7 @@ class TaskExecutorLocalStateStoresManagerTest {
config.set(CheckpointingOptions.LOCAL_RECOVERY_TASK_MANAGER_STATE_ROOT_DIRS, rootDirString);
// test configuration of the local state mode
- config.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ config.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
final WorkingDirectory workingDirectory =
WORKING_DIRECTORY_EXTENSION_WRAPPER
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
index 512e5e0a663..b8d3b0dd6a8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRecoveryTest.java
@@ -20,8 +20,8 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -69,7 +69,7 @@ class TaskExecutorRecoveryTest {
final Configuration configuration = new Configuration();
configuration.set(TaskManagerOptions.NUM_TASK_SLOTS, 2);
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
final TestingResourceManagerGateway testingResourceManagerGateway =
new TestingResourceManagerGateway();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtilsTest.java
index 563cf06b9af..09db6b5ce24 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtilsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SlotSelectionStrategyUtilsTest.java
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.util;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobmaster.slotpool.LocationPreferenceSlotSelectionStrategy;
import org.apache.flink.runtime.jobmaster.slotpool.PreviousAllocationSlotSelectionStrategy;
@@ -35,7 +35,7 @@ class SlotSelectionStrategyUtilsTest {
@Test
void testCreatePreviousAllocationSlotSelectionStrategyForLocalRecoveryStreamingJob() {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
final SlotSelectionStrategy slotSelectionStrategy =
SlotSelectionStrategyUtils.selectSlotSelectionStrategy(
@@ -48,7 +48,7 @@ class SlotSelectionStrategyUtilsTest {
@Test
void testCreateLocationPreferenceSlotSelectionStrategyForLocalRecoveryBatchJob() {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
final SlotSelectionStrategy slotSelectionStrategy =
SlotSelectionStrategyUtils.selectSlotSelectionStrategy(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
index d5136f2678f..99572894a0a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
@@ -38,6 +38,7 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.configuration.StateChangelogOptions;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -166,7 +167,7 @@ public class AutoRescalingITCase extends TestLogger {
config.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
// todo: local rescaling is not supported by changelog.
config.set(StateChangelogOptions.ENABLE_STATE_CHANGE_LOG, false);
- config.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ config.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
config.set(
CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir.toURI().toString());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogLocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogLocalRecoveryITCase.java
index a16789021de..a0790e23248 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogLocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ChangelogLocalRecoveryITCase.java
@@ -54,10 +54,10 @@ import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
-import static org.apache.flink.configuration.CheckpointingOptions.LOCAL_RECOVERY;
import static org.apache.flink.configuration.ClusterOptions.JOB_MANAGER_PROCESS_WORKING_DIR_BASE;
import static org.apache.flink.configuration.ClusterOptions.PROCESS_WORKING_DIR_BASE;
import static org.apache.flink.configuration.ClusterOptions.TASK_MANAGER_PROCESS_WORKING_DIR_BASE;
+import static org.apache.flink.configuration.StateRecoveryOptions.LOCAL_RECOVERY;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
import static org.apache.flink.test.checkpointing.ChangelogRecoveryITCaseBase.getAllStateHandleId;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
index c42b255b1fd..a333c5a77ab 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
@@ -18,8 +18,8 @@
package org.apache.flink.test.checkpointing;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.util.TestLogger;
import org.junit.Ignore;
@@ -116,7 +116,7 @@ public class LocalRecoveryITCase extends TestLogger {
protected Configuration createClusterConfig() throws IOException {
Configuration config = super.createClusterConfig();
- config.set(CheckpointingOptions.LOCAL_RECOVERY, localRecoveryEnabled);
+ config.set(StateRecoveryOptions.LOCAL_RECOVERY, localRecoveryEnabled);
return config;
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/NotifyCheckpointAbortedITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/NotifyCheckpointAbortedITCase.java
index 8bcc80293a8..811cfabd773 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/NotifyCheckpointAbortedITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/NotifyCheckpointAbortedITCase.java
@@ -27,10 +27,10 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -114,7 +114,7 @@ public class NotifyCheckpointAbortedITCase extends TestLogger {
@Before
public void setup() throws Exception {
Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
configuration.set(HighAvailabilityOptions.HA_MODE, TestingHAFactory.class.getName());
checkpointPath = new Path(TEMPORARY_FOLDER.newFolder().toURI());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointManuallyITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointManuallyITCase.java
index 81a807185a7..4f087a16ba2 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointManuallyITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointManuallyITCase.java
@@ -30,6 +30,7 @@ import org.apache.flink.changelog.fs.FsStateChangelogStorageFactory;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.core.execution.RestoreMode;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -340,7 +341,7 @@ public class ResumeCheckpointManuallyITCase extends TestLogger {
config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir.toURI().toString());
- config.set(CheckpointingOptions.LOCAL_RECOVERY, localRecovery);
+ config.set(StateRecoveryOptions.LOCAL_RECOVERY, localRecovery);
// Configure DFS DSTL for this test as it might produce too much GC pressure if
// ChangelogStateBackend is used.
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
index 0e7fd3a995a..1f1e8fbee2a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
@@ -22,12 +22,12 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.accumulators.ListAccumulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HeartbeatManagerOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint;
@@ -88,7 +88,7 @@ class LocalRecoveryITCase {
configuration.set(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, 1000L);
configuration.set(HeartbeatManagerOptions.HEARTBEAT_RPC_FAILURE_THRESHOLD, 1);
configuration.set(ClusterOptions.PROCESS_WORKING_DIR_BASE, tmpDirectory.getAbsolutePath());
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
configuration.set(TaskManagerOptions.SLOT_TIMEOUT, Duration.ofSeconds(30L));
final int parallelism = 3;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
index 44db8abdec6..b6629f5f1ee 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/DefaultSchedulerLocalRecoveryITCase.java
@@ -22,10 +22,10 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.client.program.MiniClusterClient;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
@@ -73,7 +73,7 @@ public class DefaultSchedulerLocalRecoveryITCase extends TestLogger {
private void testLocalRecoveryInternal(String failoverStrategyValue) throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
configuration.setString(EXECUTION_FAILOVER_STRATEGY.key(), failoverStrategyValue);
final int parallelism = 10;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
index 0d29bf1b0e3..001448a4dce 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/SchedulingITCase.java
@@ -22,10 +22,10 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.client.program.MiniClusterClient;
-import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -62,7 +62,7 @@ public class SchedulingITCase extends TestLogger {
@Test
public void testDisablingLocalRecovery() throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, false);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, false);
executeSchedulingTest(configuration);
}
@@ -87,7 +87,7 @@ public class SchedulingITCase extends TestLogger {
private void testLocalRecoveryInternal(String failoverStrategyValue) throws Exception {
final Configuration configuration = new Configuration();
- configuration.set(CheckpointingOptions.LOCAL_RECOVERY, true);
+ configuration.set(StateRecoveryOptions.LOCAL_RECOVERY, true);
configuration.setString(EXECUTION_FAILOVER_STRATEGY.key(), failoverStrategyValue);
executeSchedulingTest(configuration);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
index a6131a710ed..d0d9e0445d7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRecoveryCachingITCase.java
@@ -62,12 +62,12 @@ import static org.apache.flink.changelog.fs.FsStateChangelogOptions.CACHE_IDLE_T
import static org.apache.flink.changelog.fs.FsStateChangelogOptions.PREEMPTIVE_PERSIST_THRESHOLD;
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTS_DIRECTORY;
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINT_STORAGE;
-import static org.apache.flink.configuration.CheckpointingOptions.LOCAL_RECOVERY;
import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
import static org.apache.flink.configuration.RestartStrategyOptions.RESTART_STRATEGY;
import static org.apache.flink.configuration.StateBackendOptions.STATE_BACKEND;
import static org.apache.flink.configuration.StateChangelogOptions.ENABLE_STATE_CHANGE_LOG;
import static org.apache.flink.configuration.StateChangelogOptions.PERIODIC_MATERIALIZATION_ENABLED;
+import static org.apache.flink.configuration.StateRecoveryOptions.LOCAL_RECOVERY;
import static org.apache.flink.configuration.TaskManagerOptions.BUFFER_DEBLOAT_ENABLED;
import static org.apache.flink.runtime.jobgraph.SavepointRestoreSettings.forPath;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
index bacdad4a1fd..c5e05f90af1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
@@ -71,13 +71,13 @@ import static org.apache.flink.changelog.fs.FsStateChangelogOptions.PREEMPTIVE_P
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTS_DIRECTORY;
import static org.apache.flink.configuration.CheckpointingOptions.CHECKPOINT_STORAGE;
import static org.apache.flink.configuration.CheckpointingOptions.FS_SMALL_FILE_THRESHOLD;
-import static org.apache.flink.configuration.CheckpointingOptions.LOCAL_RECOVERY;
import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
import static org.apache.flink.configuration.PipelineOptions.OBJECT_REUSE;
import static org.apache.flink.configuration.RestartStrategyOptions.RESTART_STRATEGY;
import static org.apache.flink.configuration.StateBackendOptions.STATE_BACKEND;
import static org.apache.flink.configuration.StateChangelogOptions.ENABLE_STATE_CHANGE_LOG;
import static org.apache.flink.configuration.StateChangelogOptions.PERIODIC_MATERIALIZATION_INTERVAL;
+import static org.apache.flink.configuration.StateRecoveryOptions.LOCAL_RECOVERY;
import static org.apache.flink.configuration.TaskManagerOptions.BUFFER_DEBLOAT_ENABLED;
import static org.apache.flink.runtime.jobgraph.SavepointRestoreSettings.forPath;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;