From b30ab1367ae6c43b2a3696a21fa3cc7eae17b3cb Mon Sep 17 00:00:00 2001 From: Aaron He Date: Fri, 4 Sep 2026 11:57:27 -0700 Subject: [PATCH 1/4] [FLINK-40559][container] Register --claimMode/--restoreMode in application mode StandaloneApplicationClusterConfigurationParserFactory.getOptions() did not register SAVEPOINT_CLAIM_MODE or SAVEPOINT_RESTORE_MODE. Because the parser runs with stopAtNonOption=true, an unrecognized --claimMode did not fail: it ended option parsing and the remainder was swallowed into getArgs(). The claim mode was therefore unreachable in application mode and fell back to NO_CLAIM. Register both options so that createSavepointRestoreSettings can read them. Behavior is unchanged for jobs that pass neither option. Generated-by: GitHub Copilot CLI 1.0.80 --- ...tionClusterConfigurationParserFactory.java | 2 + ...ClusterConfigurationParserFactoryTest.java | 45 +++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactory.java b/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactory.java index 613050f4d000b..a7fc4ac58a573 100644 --- a/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactory.java +++ b/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactory.java @@ -81,6 +81,8 @@ public Options getOptions() { options.addOption(DYNAMIC_PROPERTY_OPTION); options.addOption(CliFrontendParser.SAVEPOINT_PATH_OPTION); options.addOption(CliFrontendParser.SAVEPOINT_ALLOW_NON_RESTORED_OPTION); + options.addOption(CliFrontendParser.SAVEPOINT_CLAIM_MODE); + options.addOption(CliFrontendParser.SAVEPOINT_RESTORE_MODE); return options; } diff --git a/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java b/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java index 1c2cb1a708d5a..e71a03df988b6 100644 --- a/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java +++ b/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java @@ -25,6 +25,7 @@ import org.apache.flink.configuration.JobManagerOptions; import org.apache.flink.configuration.PipelineOptionsInternal; import org.apache.flink.configuration.RestOptions; +import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.runtime.entrypoint.FlinkParseException; import org.apache.flink.runtime.entrypoint.parser.CommandLineParser; import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; @@ -218,6 +219,50 @@ void testSavepointRestoreSettingsParsing() throws FlinkParseException { assertThat(savepointRestoreSettings.allowNonRestoredState()).isTrue(); } + @Test + void testClaimModeParsing() throws FlinkParseException { + final String restorePath = "foobar"; + final String[] args = { + "-c", + confDirPath, + "-j", + JOB_CLASS_NAME, + "-s", + restorePath, + "--claimMode", + RecoveryClaimMode.CLAIM.name() + }; + + final SavepointRestoreSettings savepointRestoreSettings = + commandLineParser.parse(args).getSavepointRestoreSettings(); + + assertThat(savepointRestoreSettings.getRestorePath()).isEqualTo(restorePath); + assertThat(savepointRestoreSettings.getRecoveryClaimMode()) + .isEqualTo(RecoveryClaimMode.CLAIM); + } + + @Test + void testRestoreModeParsing() throws FlinkParseException { + final String restorePath = "foobar"; + final String[] args = { + "-c", + confDirPath, + "-j", + JOB_CLASS_NAME, + "-s", + restorePath, + "--restoreMode", + RecoveryClaimMode.CLAIM.name() + }; + + final SavepointRestoreSettings savepointRestoreSettings = + commandLineParser.parse(args).getSavepointRestoreSettings(); + + assertThat(savepointRestoreSettings.getRestorePath()).isEqualTo(restorePath); + assertThat(savepointRestoreSettings.getRecoveryClaimMode()) + .isEqualTo(RecoveryClaimMode.CLAIM); + } + @Test void testSetJobIdManually() throws FlinkParseException { final JobID jobId = new JobID(); From cbc263558d2fec8ad207d219a9b1a6eb71f4ad72 Mon Sep 17 00:00:00 2001 From: Aaron He Date: Fri, 4 Sep 2026 11:57:29 -0700 Subject: [PATCH 2/4] [FLINK-40559][clients] Honor --claimMode when no savepoint path is given createSavepointRestoreSettings read the claim mode only inside the branch guarded by --fromSavepoint, so passing --claimMode without a savepoint path returned SavepointRestoreSettings.none() and discarded it. Application-mode JobManagers hit this on HA recovery, where the checkpoint comes from HA storage rather than from a command line path. Branch on option presence instead, and add SavepointRestoreSettings.forRecoveryClaimMode for a claim mode with no restore path. The parsed mode is null exactly when neither option was passed, so an explicit --claimMode NO_CLAIM stays distinguishable from no override at all. allowNonRestoredState stays null unless the flag was passed, so the new branch does not start writing execution.state-recovery.ignore-unclaimed-state where nothing was written before (FLINK-39673). SavepointRestoreSettings.fromConfiguration has the same savepoint-path gate, so the `flink run` path still drops a pathless claim mode. That asymmetry is pre-existing and left untouched here; application mode writes directly to the JobManager configuration and does not go through it. Generated-by: GitHub Copilot CLI 1.0.80 --- .../flink/client/cli/CliFrontendParser.java | 49 +++++++++++-------- .../flink/client/cli/CliFrontendRunTest.java | 42 ++++++++++++++++ .../jobgraph/SavepointRestoreSettings.java | 23 +++++++++ .../SavepointRestoreSettingsTest.java | 36 ++++++++++++++ 4 files changed, 129 insertions(+), 21 deletions(-) diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java index cc76a59337afa..010884e61c6f5 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java +++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java @@ -652,36 +652,43 @@ private static void printCustomCliOptions( } public static SavepointRestoreSettings createSavepointRestoreSettings(CommandLine commandLine) { + final Boolean allowNonRestoredState = + commandLine.hasOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION.getOpt()) + ? Boolean.TRUE + : null; + // Null exactly when neither --claimMode nor --restoreMode was passed, so it also answers + // "was a claim mode explicitly requested?". Keeping an unset value null means an explicit + // "--claimMode NO_CLAIM" stays distinguishable from no override at all. + final RecoveryClaimMode recoveryClaimMode = parseRecoveryClaimMode(commandLine); + if (commandLine.hasOption(SAVEPOINT_PATH_OPTION.getOpt())) { final String savepointPath = commandLine.getOptionValue(SAVEPOINT_PATH_OPTION.getOpt()); - final Boolean allowNonRestoredState = - commandLine.hasOption(SAVEPOINT_ALLOW_NON_RESTORED_OPTION.getOpt()) - ? Boolean.TRUE - : null; - final RecoveryClaimMode recoveryClaimMode; - if (commandLine.hasOption(SAVEPOINT_CLAIM_MODE)) { - recoveryClaimMode = - ConfigurationUtils.convertValue( - commandLine.getOptionValue(SAVEPOINT_CLAIM_MODE), - RecoveryClaimMode.class); - } else if (commandLine.hasOption(SAVEPOINT_RESTORE_MODE)) { - recoveryClaimMode = - ConfigurationUtils.convertValue( - commandLine.getOptionValue(SAVEPOINT_RESTORE_MODE), - RecoveryClaimMode.class); - System.out.printf( - "The option '%s' is deprecated. Please use '%s' instead.%n", - SAVEPOINT_RESTORE_MODE.getLongOpt(), SAVEPOINT_CLAIM_MODE.getLongOpt()); - } else { - recoveryClaimMode = null; - } return SavepointRestoreSettings.forPath( savepointPath, allowNonRestoredState, recoveryClaimMode); + } else if (recoveryClaimMode != null) { + return SavepointRestoreSettings.forRecoveryClaimMode( + allowNonRestoredState, recoveryClaimMode); } else { return SavepointRestoreSettings.none(); } } + @Nullable + private static RecoveryClaimMode parseRecoveryClaimMode(CommandLine commandLine) { + if (commandLine.hasOption(SAVEPOINT_CLAIM_MODE)) { + return ConfigurationUtils.convertValue( + commandLine.getOptionValue(SAVEPOINT_CLAIM_MODE), RecoveryClaimMode.class); + } + if (commandLine.hasOption(SAVEPOINT_RESTORE_MODE)) { + System.out.printf( + "The option '%s' is deprecated. Please use '%s' instead.%n", + SAVEPOINT_RESTORE_MODE.getLongOpt(), SAVEPOINT_CLAIM_MODE.getLongOpt()); + return ConfigurationUtils.convertValue( + commandLine.getOptionValue(SAVEPOINT_RESTORE_MODE), RecoveryClaimMode.class); + } + return null; + } + // -------------------------------------------------------------------------------------------- // Line Parsing // -------------------------------------------------------------------------------------------- diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java index f1d7ac162e7f5..6b401fdf5ef6d 100644 --- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java +++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java @@ -26,6 +26,7 @@ import org.apache.flink.client.program.TestingClusterClient; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.CoreOptions; +import org.apache.flink.configuration.StateRecoveryOptions; import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; @@ -183,6 +184,47 @@ private void testRestoreMode(String flag, String arg, RecoveryClaimMode expected assertThat(savepointSettings.allowNonRestoredState()).isTrue(); } + @Test + void testClaimRestoreModeParsingWithoutSavepoint() throws Exception { + testRestoreModeWithoutSavepoint("--claimMode", "claim", RecoveryClaimMode.CLAIM); + } + + @Test + void testNoClaimRestoreModeParsingWithoutSavepoint() throws Exception { + testRestoreModeWithoutSavepoint("-rm", "no_claim", RecoveryClaimMode.NO_CLAIM); + } + + @Test + void testNoRestoreModeAndNoSavepointParsing() throws Exception { + String[] parameters = {getTestJarPath()}; + CommandLine commandLine = + CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, parameters, true); + + assertThat(CliFrontendParser.createSavepointRestoreSettings(commandLine)) + .isEqualTo(SavepointRestoreSettings.none()); + } + + private void testRestoreModeWithoutSavepoint( + String flag, String arg, RecoveryClaimMode expectedMode) throws Exception { + String[] parameters = {flag, arg, getTestJarPath()}; + CommandLine commandLine = + CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, parameters, true); + + SavepointRestoreSettings savepointSettings = + CliFrontendParser.createSavepointRestoreSettings(commandLine); + + assertThat(savepointSettings.restoreSavepoint()).isFalse(); + assertThat(savepointSettings.getRestorePath()).isNull(); + assertThat(savepointSettings.getRecoveryClaimMode()).isEqualTo(expectedMode); + + // containsKey rather than get: an explicitly requested mode must be written even when it + // equals the default, which is what distinguishes it from no override at all. + Configuration configuration = new Configuration(); + SavepointRestoreSettings.toConfiguration(savepointSettings, configuration); + assertThat(configuration.containsKey(StateRecoveryOptions.RESTORE_MODE.key())).isTrue(); + assertThat(configuration.get(StateRecoveryOptions.RESTORE_MODE)).isEqualTo(expectedMode); + } + @Test void testUnrecognizedOption() { assertThatThrownBy( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java index 20458bb48977a..81747e28327b4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java @@ -148,6 +148,13 @@ public String toString() { + ", recoveryClaimMode=" + getRecoveryClaimMode() + ')'; + } else if (recoveryClaimMode != null) { + return "SavepointRestoreSettings.forRecoveryClaimMode(" + + "allowNonRestoredState=" + + allowNonRestoredState() + + ", recoveryClaimMode=" + + getRecoveryClaimMode() + + ')'; } else { return "SavepointRestoreSettings.none()"; } @@ -189,6 +196,22 @@ public static SavepointRestoreSettings forPath( savepointPath, allowNonRestoredState, recoveryClaimMode); } + /** + * Creates restore settings that carry an explicitly requested {@link RecoveryClaimMode} without + * a savepoint restore path, as used when {@code --claimMode} is passed but {@code + * --fromSavepoint} is not. Such a claim mode would otherwise have to be represented as {@link + * #none()}, which discards it. + * + * @param allowNonRestoredState whether to allow non-restored state, or {@code null} if not + * explicitly set by the user. + * @param recoveryClaimMode how to restore from the checkpoint. + */ + public static SavepointRestoreSettings forRecoveryClaimMode( + @Nullable Boolean allowNonRestoredState, @Nonnull RecoveryClaimMode recoveryClaimMode) { + checkNotNull(recoveryClaimMode, "Recovery claim mode."); + return new SavepointRestoreSettings(null, allowNonRestoredState, recoveryClaimMode); + } + // -------------------------- Parsing to and from a configuration object // ------------------------------------ diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java index aad413603ff74..72014893430ca 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java @@ -106,6 +106,42 @@ void testNoneSettingsDoNotWriteAnything() { .isFalse(); } + @Test + void testForRecoveryClaimModeWritesClaimModeWithoutSavepointPath() { + SavepointRestoreSettings settings = + SavepointRestoreSettings.forRecoveryClaimMode(null, RecoveryClaimMode.CLAIM); + + assertThat(settings.restoreSavepoint()).isFalse(); + assertThat(settings.getRestorePath()).isNull(); + assertThat(settings.getRecoveryClaimMode()).isEqualTo(RecoveryClaimMode.CLAIM); + + Configuration configuration = new Configuration(); + SavepointRestoreSettings.toConfiguration(settings, configuration); + + assertThat(configuration.get(StateRecoveryOptions.RESTORE_MODE)) + .isEqualTo(RecoveryClaimMode.CLAIM); + assertThat(configuration.containsKey(StateRecoveryOptions.SAVEPOINT_PATH.key())).isFalse(); + // Not explicitly set, so it must not be written. + assertThat( + configuration.containsKey( + StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE.key())) + .isFalse(); + } + + @Test + void testForRecoveryClaimModeWritesExplicitlySetAllowNonRestoredState() { + SavepointRestoreSettings settings = + SavepointRestoreSettings.forRecoveryClaimMode(true, RecoveryClaimMode.CLAIM); + + Configuration configuration = new Configuration(); + SavepointRestoreSettings.toConfiguration(settings, configuration); + + assertThat(configuration.get(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE)) + .isTrue(); + assertThat(configuration.get(StateRecoveryOptions.RESTORE_MODE)) + .isEqualTo(RecoveryClaimMode.CLAIM); + } + @Test void testFromConfigurationWithAllValuesSet() { Configuration configuration = new Configuration(); From 920da525641211406e23d51eba50edfdfac932c9 Mon Sep 17 00:00:00 2001 From: Aaron He Date: Fri, 4 Sep 2026 22:26:08 -0700 Subject: [PATCH 3/4] [FLINK-40559][container] Parameterize claim mode parsing test Generated-by: GitHub Copilot CLI 1.0.83-5 --- ...ClusterConfigurationParserFactoryTest.java | 31 ++++--------------- 1 file changed, 6 insertions(+), 25 deletions(-) diff --git a/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java b/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java index e71a03df988b6..49bf928b0e180 100644 --- a/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java +++ b/flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterConfigurationParserFactoryTest.java @@ -34,6 +34,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.File; import java.io.IOException; @@ -219,30 +221,9 @@ void testSavepointRestoreSettingsParsing() throws FlinkParseException { assertThat(savepointRestoreSettings.allowNonRestoredState()).isTrue(); } - @Test - void testClaimModeParsing() throws FlinkParseException { - final String restorePath = "foobar"; - final String[] args = { - "-c", - confDirPath, - "-j", - JOB_CLASS_NAME, - "-s", - restorePath, - "--claimMode", - RecoveryClaimMode.CLAIM.name() - }; - - final SavepointRestoreSettings savepointRestoreSettings = - commandLineParser.parse(args).getSavepointRestoreSettings(); - - assertThat(savepointRestoreSettings.getRestorePath()).isEqualTo(restorePath); - assertThat(savepointRestoreSettings.getRecoveryClaimMode()) - .isEqualTo(RecoveryClaimMode.CLAIM); - } - - @Test - void testRestoreModeParsing() throws FlinkParseException { + @ParameterizedTest + @ValueSource(strings = {"--claimMode", "--restoreMode"}) + void testClaimModeParsing(String option) throws FlinkParseException { final String restorePath = "foobar"; final String[] args = { "-c", @@ -251,7 +232,7 @@ void testRestoreModeParsing() throws FlinkParseException { JOB_CLASS_NAME, "-s", restorePath, - "--restoreMode", + option, RecoveryClaimMode.CLAIM.name() }; From 05ad05a7cdfc7ed253361b88db9c2aedf80946de Mon Sep 17 00:00:00 2001 From: Aaron He Date: Tue, 15 Sep 2026 16:41:57 -0700 Subject: [PATCH 4/4] [FLINK-40559][runtime] Preserve pathless claim modes during settings reconstruction Generated-by: GitHub Copilot CLI 1.0.84-5 --- .../flink/client/cli/CliFrontendRunTest.java | 33 +++++++++- .../jobgraph/SavepointRestoreSettings.java | 11 ++-- .../SavepointRestoreSettingsTest.java | 66 ++++++++++++++++++- .../api/graph/StreamGraphGeneratorTest.java | 20 ++++++ 4 files changed, 124 insertions(+), 6 deletions(-) diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java index 6b401fdf5ef6d..4ea54857918e3 100644 --- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java +++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java @@ -22,6 +22,7 @@ import org.apache.flink.client.cli.util.DummyClusterClientServiceLoader; import org.apache.flink.client.deployment.ClusterClientServiceLoader; import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader; +import org.apache.flink.client.deployment.executors.PipelineExecutorUtils; import org.apache.flink.client.program.PackagedProgram; import org.apache.flink.client.program.TestingClusterClient; import org.apache.flink.configuration.Configuration; @@ -29,6 +30,9 @@ import org.apache.flink.configuration.StateRecoveryOptions; import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; +import org.apache.flink.streaming.api.environment.CheckpointConfig; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamGraphGenerator; import org.apache.commons.cli.CommandLine; import org.junit.jupiter.api.AfterAll; @@ -36,6 +40,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; import java.util.Arrays; @@ -211,7 +216,9 @@ private void testRestoreModeWithoutSavepoint( CliFrontendParser.parse(CliFrontendParser.RUN_OPTIONS, parameters, true); SavepointRestoreSettings savepointSettings = - CliFrontendParser.createSavepointRestoreSettings(commandLine); + ExecutionConfigAccessor.fromProgramOptions( + ProgramOptions.create(commandLine), Collections.emptyList()) + .getSavepointRestoreSettings(); assertThat(savepointSettings.restoreSavepoint()).isFalse(); assertThat(savepointSettings.getRestorePath()).isNull(); @@ -225,6 +232,30 @@ private void testRestoreModeWithoutSavepoint( assertThat(configuration.get(StateRecoveryOptions.RESTORE_MODE)).isEqualTo(expectedMode); } + @ParameterizedTest + @EnumSource(RecoveryClaimMode.class) + void testPathlessRecoveryClaimModeThroughApplicationGraph(RecoveryClaimMode mode) + throws Exception { + Configuration configuration = new Configuration(); + SavepointRestoreSettings expected = + SavepointRestoreSettings.forRecoveryClaimMode(null, mode); + SavepointRestoreSettings.toConfiguration(expected, configuration); + + StreamGraph graph = + new StreamGraphGenerator( + Collections.emptyList(), + new ExecutionConfig(), + new CheckpointConfig(), + configuration) + .generate(); + + assertThat(graph.getSavepointRestoreSettings()).isEqualTo(expected); + assertThat( + PipelineExecutorUtils.getStreamGraph(graph, configuration) + .getSavepointRestoreSettings()) + .isEqualTo(expected); + } + @Test void testUnrecognizedOption() { assertThatThrownBy( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java index 81747e28327b4..3cd10465c5971 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettings.java @@ -235,15 +235,18 @@ public static void toConfiguration( public static SavepointRestoreSettings fromConfiguration(final ReadableConfig configuration) { final String savepointPath = configuration.get(StateRecoveryOptions.SAVEPOINT_PATH); - if (savepointPath == null) { + final RecoveryClaimMode recoveryClaimMode = + configuration.getOptional(StateRecoveryOptions.RESTORE_MODE).orElse(null); + if (savepointPath == null && recoveryClaimMode == null) { return SavepointRestoreSettings.none(); } final Boolean allowNonRestored = configuration .getOptional(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE) .orElse(null); - final RecoveryClaimMode recoveryClaimMode = - configuration.getOptional(StateRecoveryOptions.RESTORE_MODE).orElse(null); - return SavepointRestoreSettings.forPath(savepointPath, allowNonRestored, recoveryClaimMode); + return savepointPath == null + ? SavepointRestoreSettings.forRecoveryClaimMode(allowNonRestored, recoveryClaimMode) + : SavepointRestoreSettings.forPath( + savepointPath, allowNonRestored, recoveryClaimMode); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java index 72014893430ca..b24a51b80ac4f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/SavepointRestoreSettingsTest.java @@ -23,6 +23,8 @@ import org.apache.flink.core.execution.RecoveryClaimMode; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -167,7 +169,69 @@ void testFromConfigurationWithNoPath() { SavepointRestoreSettings.fromConfiguration(configuration); assertThat(settings.restoreSavepoint()).isFalse(); - assertThat(settings).isEqualTo(SavepointRestoreSettings.none()); + assertThat(settings) + .isEqualTo( + SavepointRestoreSettings.forRecoveryClaimMode( + null, RecoveryClaimMode.CLAIM)); + } + + @ParameterizedTest + @CsvSource({ + ",,", + ",,false", + ",,true", + ",NO_CLAIM,", + ",NO_CLAIM,false", + ",NO_CLAIM,true", + ",CLAIM,", + ",CLAIM,false", + ",CLAIM,true", + ",LEGACY,", + ",LEGACY,false", + ",LEGACY,true", + "savepoint,,", + "savepoint,CLAIM,false", + "savepoint,LEGACY,true" + }) + void testConfigurationRoundTrip(String path, RecoveryClaimMode mode, Boolean allowNonRestored) { + Configuration input = new Configuration(); + if (path != null) { + input.set(StateRecoveryOptions.SAVEPOINT_PATH, path); + } + if (mode != null) { + input.set(StateRecoveryOptions.RESTORE_MODE, mode); + } + if (allowNonRestored != null) { + input.set(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE, allowNonRestored); + } + + Configuration output = new Configuration(); + SavepointRestoreSettings.toConfiguration( + SavepointRestoreSettings.fromConfiguration(input), output); + + if (path == null && mode == null) { + assertThat(output.toMap()).isEmpty(); + } else { + assertThat(output.toMap()).isEqualTo(input.toMap()); + } + } + + @ParameterizedTest + @CsvSource({"NO_CLAIM,", "CLAIM,", "LEGACY,", "CLAIM,NO_CLAIM", "LEGACY,CLAIM"}) + void testDeprecatedRestoreModeAlias(RecoveryClaimMode alias, RecoveryClaimMode canonical) { + Configuration input = new Configuration(); + input.setString("execution.savepoint-restore-mode", alias.name()); + if (canonical != null) { + input.set(StateRecoveryOptions.RESTORE_MODE, canonical); + } + + Configuration output = new Configuration(); + SavepointRestoreSettings.toConfiguration( + SavepointRestoreSettings.fromConfiguration(input), output); + + Configuration expected = new Configuration(); + expected.set(StateRecoveryOptions.RESTORE_MODE, canonical == null ? alias : canonical); + assertThat(output.toMap()).isEqualTo(expected.toMap()); } @Test diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java index 6384cc60f1912..5319b924d7b35 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java @@ -30,6 +30,7 @@ import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.StateRecoveryOptions; +import org.apache.flink.core.execution.RecoveryClaimMode; import org.apache.flink.core.memory.ManagedMemoryUseCase; import org.apache.flink.runtime.clusterframework.types.ResourceProfile; import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings; @@ -89,6 +90,7 @@ import org.hamcrest.TypeSafeMatcher; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; import java.util.ArrayList; @@ -739,6 +741,24 @@ void testSettingSavepointRestoreSettingsSetterOverrides() { .isEqualTo(SavepointRestoreSettings.forPath("/tmp/savepoint1")); } + @ParameterizedTest + @EnumSource(RecoveryClaimMode.class) + void testSettingPathlessRecoveryClaimMode(RecoveryClaimMode mode) { + Configuration config = new Configuration(); + config.set(StateRecoveryOptions.RESTORE_MODE, mode); + + final StreamGraph streamGraph = + new StreamGraphGenerator( + Collections.emptyList(), + new ExecutionConfig(), + new CheckpointConfig(), + config) + .generate(); + + assertThat(streamGraph.getSavepointRestoreSettings()) + .isEqualTo(SavepointRestoreSettings.forRecoveryClaimMode(null, mode)); + } + @Test void testConfigureSlotSharingGroupResource() { final SlotSharingGroup ssg1 =