(flink) 03/03: [FLINK-34190][checkpoint][cli] Deprecation message of RestoreMode#LEGACY in CLI

[hangxiang]

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 bed79d33fb7e408768d45b515d4bdee5246a3826
Author: Zakelly <zakelly....@gmail.com>
AuthorDate: Wed Jan 24 00:25:43 2024 +0800

    [FLINK-34190][checkpoint][cli] Deprecation message of RestoreMode#LEGACY in 
CLI
---
 .../java/org/apache/flink/client/cli/CliFrontendParser.java   |  4 ++--
 .../main/java/org/apache/flink/client/cli/ProgramOptions.java |  7 +++++++
 .../entrypoint/StandaloneApplicationClusterEntryPoint.java    | 11 +++++++++++
 3 files changed, 20 insertions(+), 2 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 ceca4968d21..e77765a30c2 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
@@ -141,8 +141,8 @@ public class CliFrontendParser {
                             + "[claim - claim ownership of the savepoint and 
delete once it is"
                             + " subsumed, no_claim (default) - do not claim 
ownership, the first"
                             + " checkpoint will not reuse any files from the 
restored one, legacy "
-                            + "- the old behaviour, do not assume ownership of 
the savepoint files,"
-                            + " but can reuse some shared files.");
+                            + "(deprecated) - the old behaviour, do not assume 
ownership of the "
+                            + "savepoint files, but can reuse some shared 
files.");
 
     static final Option SAVEPOINT_DISPOSE_OPTION =
             new Option("d", "dispose", true, "Path of savepoint to dispose.");
diff --git 
a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java 
b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
index 83bd563ebe5..67f9bed0309 100644
--- 
a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
+++ 
b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.runtime.jobgraph.RestoreMode;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 
 import org.apache.commons.cli.CommandLine;
@@ -138,6 +139,12 @@ public class ProgramOptions extends CommandLineOptions {
         if (getJarFilePath() == null) {
             throw new CliArgsException("Java program should be specified a JAR 
file.");
         }
+        if (savepointSettings.getRestoreMode().equals(RestoreMode.LEGACY)) {
+            System.out.printf(
+                    "Warning: The %s restore mode is deprecated, please use %s 
or"
+                            + " %s mode instead.%n",
+                    RestoreMode.LEGACY, RestoreMode.CLAIM, 
RestoreMode.NO_CLAIM);
+        }
     }
 
     public String getJarFilePath() {
diff --git 
a/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
 
b/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
index 967650bda86..795d69f683b 100644
--- 
a/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
+++ 
b/flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneApplicationClusterEntryPoint.java
@@ -29,6 +29,7 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.PipelineOptionsInternal;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils;
+import org.apache.flink.runtime.jobgraph.RestoreMode;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import 
org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
 import org.apache.flink.runtime.util.EnvironmentInformation;
@@ -61,6 +62,16 @@ public final class StandaloneApplicationClusterEntryPoint 
extends ApplicationClu
                         StandaloneApplicationClusterEntryPoint.class);
 
         Configuration configuration = 
loadConfigurationFromClusterConfig(clusterConfiguration);
+        if (clusterConfiguration
+                .getSavepointRestoreSettings()
+                .getRestoreMode()
+                .equals(RestoreMode.LEGACY)) {
+            LOG.warn(
+                    "The {} restore mode is deprecated, please use {} or {} 
mode instead.",
+                    RestoreMode.LEGACY,
+                    RestoreMode.CLAIM,
+                    RestoreMode.NO_CLAIM);
+        }
         PackagedProgram program = null;
         try {
             program = getPackagedProgram(clusterConfiguration, configuration);