[flink-kubernetes-operator] branch main updated: [FLINK-27154] Disable web.cancel.enable for application clusters

[gyfora]

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

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


The following commit(s) were added to refs/heads/main by this push:
     new b872111  [FLINK-27154] Disable web.cancel.enable for application 
clusters
b872111 is described below

commit b87211168665a34dff15cd28e3a0c0bb34f9283e
Author: Nicholas Jiang <programg...@163.com>
AuthorDate: Tue Apr 12 23:13:45 2022 +0800

    [FLINK-27154] Disable web.cancel.enable for application clusters
---
 .../operator/utils/FlinkConfigBuilder.java         | 27 ++++++++++++++--------
 .../operator/utils/FlinkConfigBuilderTest.java     |  2 ++
 2 files changed, 20 insertions(+), 9 deletions(-)

diff --git 
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilder.java
 
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilder.java
index e84ca4b..d5d8e9c 100644
--- 
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilder.java
+++ 
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilder.java
@@ -47,6 +47,7 @@ import java.time.Duration;
 import java.util.Collections;
 
 import static 
org.apache.flink.configuration.DeploymentOptionsInternal.CONF_DIR;
+import static org.apache.flink.configuration.WebOptions.CANCEL_ENABLE;
 import static 
org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE;
 import static 
org.apache.flink.kubernetes.operator.utils.FlinkUtils.mergePodTemplates;
 import static 
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
@@ -100,16 +101,24 @@ public class FlinkConfigBuilder {
                     KubernetesConfigOptions.ServiceExposedType.ClusterIP);
         }
 
-        // With last-state upgrade mode, set the default value of 
'execution.checkpointing.interval'
-        // to 5 minutes when HA is enabled.
-        if (spec.getJob() != null
-                && spec.getJob().getUpgradeMode() == UpgradeMode.LAST_STATE
-                && !effectiveConfig.contains(
-                        ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL)) 
{
-            effectiveConfig.set(
-                    ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL,
-                    DEFAULT_CHECKPOINTING_INTERVAL);
+        if (spec.getJob() != null) {
+            if (!effectiveConfig.contains(CANCEL_ENABLE)) {
+                // Set 'web.cancel.enable' to false for application 
deployments to avoid users
+                // accidentally cancelling jobs.
+                effectiveConfig.set(CANCEL_ENABLE, false);
+            }
+            // With last-state upgrade mode, set the default value of
+            // 'execution.checkpointing.interval'
+            // to 5 minutes when HA is enabled.
+            if (spec.getJob().getUpgradeMode() == UpgradeMode.LAST_STATE
+                    && !effectiveConfig.contains(
+                            
ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL)) {
+                effectiveConfig.set(
+                        ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL,
+                        DEFAULT_CHECKPOINTING_INTERVAL);
+            }
         }
+
         return this;
     }
 
diff --git 
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilderTest.java
 
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilderTest.java
index c81f7ce..b4da59b 100644
--- 
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilderTest.java
+++ 
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkConfigBuilderTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.MemorySize;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.configuration.WebOptions;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.configuration.KubernetesDeploymentTarget;
 import 
org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory;
@@ -117,6 +118,7 @@ public class FlinkConfigBuilderTest {
         Assert.assertEquals(
                 KubernetesConfigOptions.ServiceExposedType.ClusterIP,
                 
configuration.get(KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE));
+        Assert.assertEquals(false, 
configuration.get(WebOptions.CANCEL_ENABLE));
 
         FlinkDeployment deployment = 
ReconciliationUtils.clone(flinkDeployment);
         deployment