From 60654f4f50ba98d482729ba19463986703881e05 Mon Sep 17 00:00:00 2001
From: Nicholas Jiang <programgeek@163.com>
Date: Wed, 11 May 2022 17:14:40 +0800
Subject: [PATCH] [FLINK-27261] Disable 'web.cancel.enable' for session cluster

This closes #198.
---
 .../kubernetes/operator/config/FlinkConfigBuilder.java     | 6 ++----
 .../kubernetes/operator/config/FlinkConfigBuilderTest.java | 7 +++++++
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
index 727c4377c2..15f25843c1 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
@@ -134,12 +134,10 @@ protected FlinkConfigBuilder applyFlinkConfiguration() {
         // Adapt default rest service type from 1.15+
         setDefaultConf(
                 REST_SERVICE_EXPOSED_TYPE, KubernetesConfigOptions.ServiceExposedType.ClusterIP);
+        // Set 'web.cancel.enable' to false to avoid users accidentally cancelling jobs.
+        setDefaultConf(CANCEL_ENABLE, false);
 
         if (spec.getJob() != null) {
-            // Set 'web.cancel.enable' to false for application deployments to avoid users
-            // accidentally cancelling jobs.
-            setDefaultConf(CANCEL_ENABLE, false);
-
             // With last-state upgrade mode, set the default value of
             // 'execution.checkpointing.interval'
             // to 5 minutes when HA is enabled.
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
index d26d455cd9..2cc456cf80 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
@@ -146,6 +146,13 @@ public void testApplyFlinkConfiguration() {
         Assertions.assertEquals(
                 DEFAULT_CHECKPOINTING_INTERVAL,
                 configuration.get(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL));
+
+        deployment = TestUtils.buildSessionCluster();
+        configuration =
+                new FlinkConfigBuilder(deployment, new Configuration())
+                        .applyFlinkConfiguration()
+                        .build();
+        Assertions.assertEquals(false, configuration.get(WebOptions.CANCEL_ENABLE));
     }
 
     @Test