[FLINK-38915][Kubernetes Operator] In-place suspension for FlinkBlueGreenDeployment

.github/workflows/docs.yaml

@@ -82,7 +82,7 @@ jobs:
           -Dcheckstyle.skip=true \
           -Dspotless.check.skip=true \
           -Denforcer.skip=true \
-          -Dheader="<a href=\"http://flink.apache.org/\" target=\"_top\"><h1>Back to Flink Website</h1></a> <script>var _paq=window._paq=window._paq||[];_paq.push([\"disableCookies\"]),_paq.push([\"setDomains\",[\"*.flink.apache.org\",\"*.nightlies.apache.org/flink\"]]),_paq.push([\"trackPageView\"]),_paq.push([\"enableLinkTracking\"]),function(){var u=\"//matomo.privacy.apache.org/\";_paq.push([\"setTrackerUrl\",u+\"matomo.php\"]),_paq.push([\"setSiteId\",\"1\"]);var d=document, g=d.createElement('script'), s=d.getElementsByTagName('script')[0];g.async=true; g.src=u+'matomo.js'; s.parentNode.insertBefore(g,s)}();</script>"
+          -Dheader="<a href=\"http://flink.apache.org/\" target=\"_top\"><h1>Back to Flink Website</h1></a> <script>var _paq=window._paq=window._paq||[];_paq.push([\"disableCookies\"]),_paq.push([\"setDomains\",[\"*.flink.apache.org\",\"*.nightlies.apache.org/flink\"]]),_paq.push([\"trackPageView\"]),_paq.push([\"enableLinkTracking\"]),function(){var u=\"//analytics.apache.org/\";_paq.push([\"setTrackerUrl\",u+\"matomo.php\"]),_paq.push([\"setSiteId\",\"1\"]);var d=document, g=d.createElement('script'), s=d.getElementsByTagName('script')[0];g.async=true; g.src=u+'matomo.js'; s.parentNode.insertBefore(g,s)}();</script>"
 
           mv target/site/apidocs docs/target/api/java
       - name: Upload documentation

docs/layouts/_default/baseof.html

@@ -34,7 +34,7 @@
     _paq.push(['trackPageView']);
     _paq.push(['enableLinkTracking']);
     (function() {
-      var u="//matomo.privacy.apache.org/";
+      var u="//analytics.apache.org/";
       _paq.push(['setTrackerUrl', u+'matomo.php']);
       _paq.push(['setSiteId', '1']);
       var d=document, g=d.createElement('script'), s=d.getElementsByTagName('script')[0];

docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html

@@ -212,6 +212,12 @@
             <td>Boolean</td>
             <td>Indicate whether a savepoint must be taken when deleting a FlinkDeployment or FlinkSessionJob.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.job.submission.timeout</h5></td>
+            <td style="word-wrap: break-word;">10 min</td>
+            <td>Duration</td>
+            <td>The timeout for session job submissions.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.job.upgrade.ignore-pending-savepoint</h5></td>
             <td style="word-wrap: break-word;">false</td>

docs/layouts/shortcodes/generated/system_section.html

@@ -50,6 +50,12 @@
             <td>Duration</td>
             <td>The timeout for the observer to wait the flink rest client to return.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.job.submission.timeout</h5></td>
+            <td style="word-wrap: break-word;">10 min</td>
+            <td>Duration</td>
+            <td>The timeout for session job submissions.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.leader-election.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

flink-autoscaler-plugin-jdbc/pom.xml

@@ -73,6 +73,11 @@ under the License.
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
         <!-- Test dependencies -->
 
         <dependency>

flink-autoscaler-plugin-jdbc/src/main/java/org/apache/flink/autoscaler/jdbc/event/JdbcAutoScalerEventHandler.java

@@ -25,8 +25,7 @@
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
-
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 

flink-autoscaler-standalone/pom.xml

@@ -45,6 +45,11 @@ under the License.
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-runtime</artifactId>

flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerExecutor.java

@@ -26,8 +26,7 @@
 import org.apache.flink.configuration.UnmodifiableConfiguration;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
-import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
-
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;

flink-autoscaler/pom.xml

@@ -78,6 +78,11 @@ under the License.
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-test-utils-junit</artifactId>

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/RestApiMetricsCollector.java

@@ -35,8 +35,7 @@
 import org.apache.flink.runtime.rest.messages.job.metrics.MetricsAggregationParameter;
 import org.apache.flink.runtime.rest.messages.job.metrics.MetricsFilterParameter;
 
-import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
-
+import com.google.common.collect.ImmutableMap;
 import lombok.SneakyThrows;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/JobTopology.java

@@ -21,14 +21,14 @@
 import org.apache.flink.runtime.instance.SlotSharingGroupId;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 
-import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
-import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java

@@ -22,13 +22,16 @@
 import org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.fabric8.crd.generator.annotation.PrinterColumn;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 import org.apache.commons.lang3.StringUtils;
 
+import static org.apache.flink.api.common.JobStatus.RECONCILING;
+
 /** Last observed common status of the Flink deployment/Flink SessionJob. */
 @Experimental
 @Data
@@ -121,4 +124,13 @@ public ResourceLifecycleState getLifecycleState() {
 
         return ResourceLifecycleState.DEPLOYED;
     }
+
+    @JsonIgnore
+    public boolean isJobCancellable() {
+        var jobState = jobStatus.getState();
+        if (jobState == null) {
+            return false;
+        }
+        return RECONCILING != jobState && !jobState.isGloballyTerminalState();
+    }
 }

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java

@@ -20,6 +20,7 @@
 import org.apache.flink.annotation.Experimental;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import lombok.AllArgsConstructor;
 import lombok.Data;
@@ -55,4 +56,10 @@ public class FlinkDeploymentStatus extends CommonStatus<FlinkDeploymentSpec> {
 
     /** Information about the TaskManagers for the scale subresource. */
     private TaskManagerInfo taskManager;
+
+    @JsonIgnore
+    @Override
+    public boolean isJobCancellable() {
+        return super.isJobCancellable() && jobManagerDeploymentStatus.isRestApiAvailable();
+    }
 }

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java

@@ -35,4 +35,8 @@ public enum JobManagerDeploymentStatus {
 
     /** Deployment in terminal error, requires spec change for reconciliation to continue. */
     ERROR;
+
+    public boolean isRestApiAvailable() {
+        return this == READY;
+    }
 }

flink-kubernetes-operator/pom.xml

@@ -154,6 +154,11 @@ under the License.
             <version>${log4j.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
         <!-- Test -->
         <dependency>
             <groupId>org.apache.flink</groupId>

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkConfMountDecorator.java

@@ -43,7 +43,7 @@
 import org.apache.flink.kubernetes.shaded.io.fabric8.kubernetes.api.model.VolumeBuilder;
 import org.apache.flink.kubernetes.utils.Constants;
 
-import org.apache.flink.shaded.guava31.com.google.common.io.Files;
+import com.google.common.io.Files;
 
 import java.io.File;
 import java.io.IOException;

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java

@@ -26,5 +26,23 @@ public enum BlueGreenDiffType {
     TRANSITION,
 
     /** Changes that only affect the child FlinkDeploymentSpec. */
-    PATCH_CHILD
+    PATCH_CHILD,
+
+    /**
+     * Full redeploy from user-specified savepoint. Triggered when savepointRedeployNonce changes.
+     * Uses the initialSavepointPath from spec instead of taking a new savepoint.
+     */
+    SAVEPOINT_REDEPLOY,
+
+    /**
+     * In-place suspension. Triggered when job.state changes from RUNNING to SUSPENDED. Suspends the
+     * currently active child without creating a new deployment.
+     */
+    SUSPEND,
+
+    /**
+     * Resume from suspension. Triggered when job.state changes from SUSPENDED to RUNNING. Spins up
+     * the child with the current (potentially updated) spec.
+     */
+    RESUME
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java

@@ -363,9 +363,11 @@ private int getParallelism() {
                     * effectiveConfig.get(TaskManagerOptions.NUM_TASK_SLOTS);
         }
 
-        Optional<Integer> maxOverrideParallelism = getMaxParallelismFromOverrideConfig();
-        if (maxOverrideParallelism.isPresent() && maxOverrideParallelism.get() > 0) {
-            return maxOverrideParallelism.get();
+        if (KubernetesDeploymentMode.STANDALONE.equals(spec.getMode())) {
+            Optional<Integer> maxOverrideParallelism = getMaxParallelismFromOverrideConfig();
+            if (maxOverrideParallelism.isPresent() && maxOverrideParallelism.get() > 0) {
+                return maxOverrideParallelism.get();
+            }
         }
 
         return spec.getJob().getParallelism();

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java

@@ -33,13 +33,12 @@
 import org.apache.flink.kubernetes.operator.utils.EnvUtils;
 import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
 
-import org.apache.flink.shaded.guava31.com.google.common.cache.Cache;
-import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
-import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
-import org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
-
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import lombok.Builder;
 import lombok.SneakyThrows;

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java

@@ -80,6 +80,7 @@ public class FlinkOperatorConfiguration {
     int reportedExceptionEventsMaxCount;
     int reportedExceptionEventsMaxStackTraceLength;
     boolean manageIngress;
+    Duration jobSubmissionTimeout;
 
     public static FlinkOperatorConfiguration fromConfiguration(Configuration operatorConfig) {
         Duration reconcileInterval =
@@ -207,6 +208,9 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
         boolean manageIngress =
                 operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_MANAGE_INGRESS);
 
+        Duration jobSubmissionTimeout =
+                operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_JOB_SUBMISSION_TIMEOUT);
+
         return new FlinkOperatorConfiguration(
                 reconcileInterval,
                 reconcilerMaxParallelism,
@@ -239,7 +243,8 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
                 slowRequestThreshold,
                 reportedExceptionEventsMaxCount,
                 reportedExceptionEventsMaxStackTraceLength,
-                manageIngress);
+                manageIngress,
+                jobSubmissionTimeout);
     }
 
     private static GenericRetry getRetryConfig(Configuration conf) {

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -134,6 +134,13 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "The timeout for the resource clean up to wait for flink to shutdown cluster.");
 
+    @Documentation.Section(SECTION_SYSTEM)
+    public static final ConfigOption<Duration> OPERATOR_JOB_SUBMISSION_TIMEOUT =
+            operatorConfig("job.submission.timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(10))
+                    .withDescription("The timeout for session job submissions.");
+
     @Documentation.Section(SECTION_DYNAMIC)
     public static final ConfigOption<Boolean> DEPLOYMENT_ROLLBACK_ENABLED =
             operatorConfig("deployment.rollback.enabled")