Merge branch 'apache:main' into fix-missing-JM-status

nishita-09 · web-flow · commit 1d3eed0a2401 · 2025-07-23T19:26:44.000+05:30
diff --git a/docs/content.zh/docs/operations/rbac.md b/docs/content.zh/docs/operations/rbac.md
@@ -85,6 +85,7 @@ For each additional namespace that runs the Flink jobs, users need to do the fol
       - apps
       resources:
       - deployments
+      - deployments/finalizers
       verbs:
       - '*'
     ---
diff --git a/docs/content/docs/operations/rbac.md b/docs/content/docs/operations/rbac.md
@@ -85,6 +85,7 @@ For each additional namespace that runs the Flink jobs, users need to do the fol
       - apps
       resources:
       - deployments
+      - deployments/finalizers
       verbs:
       - '*'
     ---
diff --git a/docs/layouts/shortcodes/generated/dynamic_section.html b/docs/layouts/shortcodes/generated/dynamic_section.html
@@ -194,6 +194,12 @@
             <td>Duration</td>
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session.block-on-unmanaged-jobs</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.snapshot.resource.enabled</h5></td>
             <td style="word-wrap: break-word;">true</td>
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -416,6 +416,12 @@
             <td>Duration</td>
             <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.operator.session.block-on-unmanaged-jobs</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.operator.snapshot.resource.enabled</h5></td>
             <td style="word-wrap: break-word;">true</td>
diff --git a/examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java b/examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java
@@ -55,7 +55,7 @@
  *
  *                    A concrete example: "1;2;4\n4;2;1"
  *                      Two branches are created with three tasks each. On the first branch, the tasks have
- *                      a load of 1, 2, and 3 respectively. On the second branch, the tasks have the load reversed.
+ *                      a load of 1, 2, and 4 respectively. On the second branch, the tasks have the load reversed.
  *                      This means, that at peak Flink Autoscaling at target utilization of 0.5, the parallelisms of
  *                      the tasks will be 2, 4, 8 for branch one and vise-versa for branch two.
  * </pre>
diff --git a/flink-kubernetes-operator-api/pom.xml b/flink-kubernetes-operator-api/pom.xml
@@ -257,6 +257,23 @@ under the License.
                             </target>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>statesnapshot-crd-compatibility-check</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <target>
+                                <java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"
+                                      fork="true" failonerror="true">
+                                    <classpath refid="maven.compile.classpath"/>
+                                    <arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
+                                    <arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.10.0/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
+                                </java>
+                            </target>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
             <plugin>
diff --git a/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityCheckerTest.java b/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityCheckerTest.java
@@ -19,6 +19,7 @@
 
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -271,6 +272,19 @@ public void testCreateFlinkDeploymentIgnoreUnknownFields() throws IOException {
         assertEquals(flinkDeploymentWithUnknownFields.toString(), flinkDeployment.toString());
     }
 
+    @Test
+    public void testCreateFlinkStateSnapshotIgnoreUnknownFields() throws IOException {
+        FlinkStateSnapshot flinkSnapshotWithUnknownFields =
+                objectMapper.readValue(
+                        new File("src/test/resources/test-snapshot-with-unknown-fields.yaml"),
+                        FlinkStateSnapshot.class);
+        FlinkStateSnapshot flinkSnapshot =
+                objectMapper.readValue(
+                        new File("src/test/resources/test-snapshot.yaml"),
+                        FlinkStateSnapshot.class);
+        assertEquals(flinkSnapshotWithUnknownFields.toString(), flinkSnapshot.toString());
+    }
+
     @Test
     public void testGetSchemaFromUrl() throws IOException {
         var fileUrl =
diff --git a/flink-kubernetes-operator-api/src/test/resources/test-snapshot-with-unknown-fields.yaml b/flink-kubernetes-operator-api/src/test/resources/test-snapshot-with-unknown-fields.yaml
@@ -0,0 +1,30 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkStateSnapshot
+metadata:
+  name: example-savepoint
+spec:
+  backoffLimit: 0
+  jobReference:
+    kind: FlinkDeployment
+    name: example-deployment
+    unknownField: testUnknownField
+  savepoint:
+    unknownField: testUnknownField
+  unknownField: testUnknownField
diff --git a/flink-kubernetes-operator-api/src/test/resources/test-snapshot.yaml b/flink-kubernetes-operator-api/src/test/resources/test-snapshot.yaml
@@ -0,0 +1,27 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkStateSnapshot
+metadata:
+  name: example-savepoint
+spec:
+  backoffLimit: 0
+  jobReference:
+    kind: FlinkDeployment
+    name: example-deployment
+  savepoint: {}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -647,6 +647,14 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "Indicate whether the job should be drained when stopping with savepoint.");
 
+    @Documentation.Section(SECTION_DYNAMIC)
+    public static final ConfigOption<Boolean> BLOCK_ON_UNMANAGED_JOBS =
+            operatorConfig("session.block-on-unmanaged-jobs")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Block FlinkDeployment deletion if unmanaged jobs (jobs not managed by FlinkSessionJob resources) are running in the session cluster. Example: Jobs submitted via CLI.");
+
     @Documentation.Section(SECTION_ADVANCED)
     public static final ConfigOption<Duration> REFRESH_CLUSTER_RESOURCE_VIEW =
             operatorConfig("cluster.resource-view.refresh-interval")
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -17,6 +17,8 @@
 
 package org.apache.flink.kubernetes.operator.reconciler.deployment;
 
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.autoscaler.NoopJobAutoscaler;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
@@ -25,11 +27,16 @@
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
 import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
 import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
 import org.apache.flink.kubernetes.operator.utils.IngressUtils;
 import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
 
 import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
 import org.slf4j.Logger;
@@ -120,6 +127,37 @@ private void recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
                 .setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
     }
 
+    // Detects jobs which are not in globally terminated states
+    @VisibleForTesting
+    Set<JobID> getNonTerminalJobs(FlinkResourceContext<FlinkDeployment> ctx) {
+        LOG.debug("Starting nonTerminal jobs detection for session cluster");
+        try {
+            // Get all jobs running in the Flink cluster
+            var flinkService = ctx.getFlinkService();
+            var clusterClient = flinkService.getClusterClient(ctx.getObserveConfig());
+            var allJobs =
+                    clusterClient
+                            .sendRequest(
+                                    JobsOverviewHeaders.getInstance(),
+                                    EmptyMessageParameters.getInstance(),
+                                    EmptyRequestBody.getInstance())
+                            .get()
+                            .getJobs();
+
+            // running job Ids
+            Set<JobID> nonTerminalJobIds =
+                    allJobs.stream()
+                            .filter(job -> !job.getStatus().isGloballyTerminalState())
+                            .map(JobDetails::getJobId)
+                            .collect(Collectors.toSet());
+
+            return nonTerminalJobIds;
+        } catch (Exception e) {
+            LOG.warn("Failed to detect nonTerminal jobs in session cluster", e);
+            return Set.of();
+        }
+    }
+
     @Override
     public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx) {
         Set<FlinkSessionJob> sessionJobs =
@@ -143,13 +181,39 @@ public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx)
             }
             return DeleteControl.noFinalizerRemoval()
                     .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
-        } else {
-            LOG.info("Stopping session cluster");
-            var conf = ctx.getDeployConfig(ctx.getResource().getSpec());
-            ctx.getFlinkService()
-                    .deleteClusterDeployment(
-                            deployment.getMetadata(), deployment.getStatus(), conf, true);
-            return DeleteControl.defaultDelete();
         }
+
+        // Check for non-terminated jobs if the option is enabled (Enabled by default) , after
+        // sessionJobs are deleted
+        boolean blockOnUnmanagedJobs =
+                ctx.getObserveConfig()
+                        .getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
+        if (blockOnUnmanagedJobs) {
+            Set<JobID> nonTerminalJobs = getNonTerminalJobs(ctx);
+            if (!nonTerminalJobs.isEmpty()) {
+                var error =
+                        String.format(
+                                "The session cluster has non terminated jobs %s that should be cancelled first",
+                                nonTerminalJobs.stream()
+                                        .map(JobID::toHexString)
+                                        .collect(Collectors.toList()));
+                eventRecorder.triggerEvent(
+                        deployment,
+                        EventRecorder.Type.Warning,
+                        EventRecorder.Reason.CleanupFailed,
+                        EventRecorder.Component.Operator,
+                        error,
+                        ctx.getKubernetesClient());
+                return DeleteControl.noFinalizerRemoval()
+                        .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+            }
+        }
+
+        LOG.info("Stopping session cluster");
+        var conf = ctx.getObserveConfig();
+        ctx.getFlinkService()
+                .deleteClusterDeployment(
+                        deployment.getMetadata(), deployment.getStatus(), conf, true);
+        return DeleteControl.defaultDelete();
     }
 }
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@`
`55`	`55`	`*`
`56`	`56`	`* A concrete example: "1;2;4\n4;2;1"`
`57`	`57`	`* Two branches are created with three tasks each. On the first branch, the tasks have`
`58`		`- * a load of 1, 2, and 3 respectively. On the second branch, the tasks have the load reversed.`
	`58`	`+ * a load of 1, 2, and 4 respectively. On the second branch, the tasks have the load reversed.`
`59`	`59`	`* This means, that at peak Flink Autoscaling at target utilization of 0.5, the parallelisms of`
`60`	`60`	`* the tasks will be 2, 4, 8 for branch one and vise-versa for branch two.`
`61`	`61`	`* </pre>`