[flink-kubernetes-operator] branch release-1.2 updated: [FLINK-29633] Pass fromSavepoint and allowNonRestoredState argument correctly

[gyfora]

This is an automated email from the ASF dual-hosted git repository.

gyfora pushed a commit to branch release-1.2
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git


The following commit(s) were added to refs/heads/release-1.2 by this push:
     new 66f2814a [FLINK-29633] Pass fromSavepoint and allowNonRestoredState 
argument correctly
66f2814a is described below

commit 66f2814a50f897031dbc8c348c03a83192378274
Author: Yaroslav Tkachenko <sapie...@gmail.com>
AuthorDate: Sat Oct 15 13:13:46 2022 -0700

    [FLINK-29633] Pass fromSavepoint and allowNonRestoredState argument 
correctly
---
 .../kubeclient/decorators/CmdStandaloneJobManagerDecorator.java  | 9 +++++++--
 .../parameters/StandaloneKubernetesJobManagerParameters.java     | 7 +++++++
 .../decorators/CmdStandaloneJobManagerDecoratorTest.java         | 6 ++++--
 3 files changed, 18 insertions(+), 4 deletions(-)

diff --git 
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
 
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
index 44ded3c2..d50a9bc2 100644
--- 
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
+++ 
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
@@ -80,9 +80,14 @@ public class CmdStandaloneJobManagerDecorator extends 
AbstractKubernetesStepDeco
         }
 
         Boolean allowNonRestoredState = 
kubernetesJobManagerParameters.getAllowNonRestoredState();
-        if (allowNonRestoredState != null) {
+        if (allowNonRestoredState != null && allowNonRestoredState) {
             args.add("--allowNonRestoredState");
-            args.add(allowNonRestoredState.toString());
+        }
+
+        String savepointPath = 
kubernetesJobManagerParameters.getSavepointPath();
+        if (savepointPath != null) {
+            args.add("--fromSavepoint");
+            args.add(savepointPath);
         }
 
         List<String> jobSpecArgs = 
kubernetesJobManagerParameters.getJobSpecArgs();
diff --git 
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
 
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
index b2ca29af..15a164c8 100644
--- 
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
+++ 
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
@@ -90,6 +90,13 @@ public class StandaloneKubernetesJobManagerParameters 
extends KubernetesJobManag
         return null;
     }
 
+    public String getSavepointPath() {
+        if (flinkConfig.contains(SavepointConfigOptions.SAVEPOINT_PATH)) {
+            return flinkConfig.get(SavepointConfigOptions.SAVEPOINT_PATH);
+        }
+        return null;
+    }
+
     public boolean isPipelineClasspathDefined() {
         return flinkConfig.contains(PipelineOptions.CLASSPATHS);
     }
diff --git 
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
 
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
index 8bdb0f33..6ffcd80b 100644
--- 
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
+++ 
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
@@ -75,7 +75,8 @@ public class CmdStandaloneJobManagerDecoratorTest {
                 StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
                 
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
         configuration.set(ApplicationConfiguration.APPLICATION_MAIN_CLASS, 
testMainClass);
-        
configuration.set(SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE, 
false);
+        
configuration.set(SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE, 
true);
+        configuration.set(SavepointConfigOptions.SAVEPOINT_PATH, 
"/tmp/savepoint/path");
 
         FlinkPod decoratedPod = decorator.decorateFlinkPod(new 
FlinkPod.Builder().build());
         assertThat(
@@ -85,7 +86,8 @@ public class CmdStandaloneJobManagerDecoratorTest {
                 containsInAnyOrder(
                         CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
                         "--allowNonRestoredState",
-                        "false",
+                        "--fromSavepoint",
+                        "/tmp/savepoint/path",
                         "--job-classname",
                         testMainClass));
     }