[FLINK-28648][Flink-Kubernetes-Operator] Changed getUnManagedJobs -> getNonTerminated + annotated @VisibleForTesting

Author: nishita-pattanayak

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java

@@ -17,6 +17,7 @@
 
 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;
@@ -130,11 +131,9 @@ private void recoverSession(FlinkResourceContext<FlinkDeployment> ctx) throws Ex
      * Detects unmanaged jobs running in the session cluster. Unmanaged jobs are jobs that exist in
      * the Flink cluster but are not managed by FlinkSessionJob resources.
      */
-    private Set<JobID> getUnmanagedJobs(
-            FlinkResourceContext<FlinkDeployment> ctx, Set<FlinkSessionJob> sessionJobs) {
-        LOG.info(
-                "Starting unmanaged job detection for session cluster: {}",
-                ctx.getResource().getMetadata().getName());
+    @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();
@@ -148,35 +147,17 @@ private Set<JobID> getUnmanagedJobs(
                             .get()
                             .getJobs();
 
-            // Get job IDs managed by FlinkSessionJob resources
-            Set<JobID> managedJobIds =
-                    sessionJobs.stream()
-                            .map(job -> job.getStatus().getJobStatus().getJobId())
-                            .filter(jobId -> jobId != null)
-                            .map(JobID::fromHexString)
-                            .collect(Collectors.toSet());
-
-            // Get job IDs that are not managed by FlinkSessionJob resources and are currently
             // running
-            //    FAILED(TerminalState.GLOBALLY)
-            //    CANCELED(TerminalState.GLOBALLY)
-            //    FINISHED(TerminalState.GLOBALLY)
             // Above terminal states are not considered since they are no longer active jobs
-            Set<JobID> unmanagedJobIds =
+            Set<JobID> nonTerminalJobIds =
                     allJobs.stream()
-                            .filter(
-                                    job ->
-                                            !job.getStatus()
-                                                    .isGloballyTerminalState()) // Only consider
-                            // running jobs
+                            .filter(job -> !job.getStatus().isGloballyTerminalState())
                             .map(JobDetails::getJobId)
-                            .filter(jobId -> !managedJobIds.contains(jobId))
                             .collect(Collectors.toSet());
 
-            LOG.info("Detected {} unmanaged job IDs: {}", unmanagedJobIds.size(), unmanagedJobIds);
-            return unmanagedJobIds;
+            return nonTerminalJobIds;
         } catch (Exception e) {
-            LOG.warn("Failed to detect unmanaged jobs in session cluster", e);
+            LOG.warn("Failed to detect nonTerminal jobs in session cluster", e);
             return Set.of();
         }
     }
@@ -206,20 +187,19 @@ public DeleteControl cleanupInternal(FlinkResourceContext<FlinkDeployment> ctx)
                     .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
         }
 
-        // Check for unmanaged jobs if the option is enabled (Enabled by default)
+        // Check for unmanaged non-terminated jobs if the option is enabled (Enabled by default)
         boolean blockOnUnmanagedJobs =
                 ctx.getObserveConfig()
                         .getBoolean(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS);
         if (blockOnUnmanagedJobs) {
-            Set<JobID> unmanagedJobs = getUnmanagedJobs(ctx, sessionJobs);
-            if (!unmanagedJobs.isEmpty()) {
+            Set<JobID> nonTerminalJobs = getNonTerminalJobs(ctx);
+            if (!nonTerminalJobs.isEmpty()) {
                 var error =
                         String.format(
-                                "The session cluster has unmanaged jobs %s that should be cancelled first",
-                                unmanagedJobs.stream()
+                                "The session cluster has non terminated jobs %s that should be cancelled first",
+                                nonTerminalJobs.stream()
                                         .map(JobID::toHexString)
                                         .collect(Collectors.toList()));
-                LOG.warn(error);
                 if (eventRecorder.triggerEvent(
                         deployment,
                         EventRecorder.Type.Warning,

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java

@@ -32,7 +32,6 @@
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
 import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
-import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
 import org.apache.flink.kubernetes.operator.reconciler.TestReconcilerAdapter;
 import org.apache.flink.runtime.client.JobStatusMessage;
 
@@ -49,7 +48,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -151,8 +149,7 @@ public void testSetOwnerReference() throws Exception {
     }
 
     @Test
-    public void testGetUnmanagedJobs() throws Exception {
-        // Create a session cluster deployment with blocking enabled
+    public void testGetNonTerminalJobs() throws Exception {
         FlinkDeployment deployment = TestUtils.buildSessionCluster();
         deployment
                 .getSpec()
@@ -166,7 +163,6 @@ public void testGetUnmanagedJobs() throws Exception {
                         .getFlinkConfiguration()
                         .get(KubernetesOperatorConfigOptions.BLOCK_ON_UNMANAGED_JOBS.key()));
 
-        // First, deploy the session cluster to set up proper state
         reconciler.reconcile(deployment, flinkService.getContext());
 
         // Verify deployment is in DEPLOYED state
@@ -181,7 +177,7 @@ public void testGetUnmanagedJobs() throws Exception {
         JobID unmanagedTerminatedJobId = new JobID();
         JobID unmanagedRunningJobId2 = new JobID();
 
-        // Add a mix of managed and unmanaged jobs to the testing service
+        // Add jobs to the testing service
         flinkService
                 .listJobs()
                 .add(
@@ -253,60 +249,59 @@ public void testGetUnmanagedJobs() throws Exception {
         sessionJobs.add(managedSessionJob1);
         sessionJobs.add(managedSessionJob2);
 
-        // Test with blocking enabled - should identify running unmanaged jobs correctly
+        // Test with blocking enabled - should identify all non-terminal jobs
         var context = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
         var resourceContext = getResourceContext(deployment, context);
 
-        // Use reflection to access the private getUnmanagedJobs method
-        var sessionReconciler = reconciler.getReconciler();
-        var getUnmanagedJobsMethod =
-                sessionReconciler
-                        .getClass()
-                        .getDeclaredMethod(
-                                "getUnmanagedJobs", FlinkResourceContext.class, Set.class);
-        getUnmanagedJobsMethod.setAccessible(true);
-
-        @SuppressWarnings("unchecked")
-        var unmanagedJobs =
-                (java.util.Set<String>)
-                        getUnmanagedJobsMethod.invoke(
-                                sessionReconciler, resourceContext, sessionJobs);
-
-        // Verify only RUNNING unmanaged jobs are identified - should be 2
-        assertEquals(2, unmanagedJobs.size(), "Should identify exactly 2 running unmanaged jobs");
-
-        // Verify that none of the managed job IDs are in the unmanaged list
-        Set<String> managedJobIds =
-                sessionJobs.stream()
-                        .map(job -> job.getStatus().getJobStatus().getJobId())
-                        .collect(Collectors.toSet());
-
-        for (String managedJobId : managedJobIds) {
-            assertFalse(
-                    unmanagedJobs.contains(managedJobId),
-                    "Managed job " + managedJobId + " should NOT be in unmanaged jobs");
-        }
+        var sessionReconciler = (SessionReconciler) reconciler.getReconciler();
+        Set<JobID> nonTerminalJobs = sessionReconciler.getNonTerminalJobs(resourceContext);
+
+        // Verify all non-terminal jobs are identified - should be 4 (2 managed + 2 unmanaged
+        // running)
+        assertEquals(4, nonTerminalJobs.size(), "Should identify exactly 4 non-terminal jobs");
+
+        assertTrue(
+                nonTerminalJobs.contains(unmanagedRunningJobId1),
+                "Should contain unmanagedRunningJobId1");
+        assertTrue(
+                nonTerminalJobs.contains(unmanagedRunningJobId2),
+                "Should contain unmanagedRunningJobId2");
 
-        // Verify managed jobs are properly set up (we should have 2 managed jobs)
-        assertEquals(2, managedJobIds.size(), "Should have 2 managed jobs configured");
+        // Verify terminated job is not included
+        assertFalse(
+                nonTerminalJobs.contains(unmanagedTerminatedJobId),
+                "Should not contain terminated job");
+
+        // Test scenario with only unmanaged jobs
+        flinkService
+                .listJobs()
+                .removeIf(
+                        job ->
+                                job.f1.getJobId().equals(managedJobId1)
+                                        || job.f1.getJobId().equals(managedJobId2));
+
+        Set<JobID> nonTerminalJobsAfterSessionJobsRemoval =
+                sessionReconciler.getNonTerminalJobs(resourceContext);
+
+        assertEquals(
+                2,
+                nonTerminalJobsAfterSessionJobsRemoval.size(),
+                "Should have 2 non-terminal jobs when sessionjobs are deleted");
 
-        // Test scenario with no running unmanaged jobs
+        // Test scenario with no running jobs
         flinkService
                 .listJobs()
                 .removeIf(
                         job ->
                                 job.f1.getJobId().equals(unmanagedRunningJobId1)
                                         || job.f1.getJobId().equals(unmanagedRunningJobId2));
 
-        @SuppressWarnings("unchecked")
-        var unmanagedJobsAfterRemoval =
-                (java.util.Set<String>)
-                        getUnmanagedJobsMethod.invoke(
-                                sessionReconciler, resourceContext, sessionJobs);
+        Set<JobID> nonTerminalJobsAfterRemoval =
+                sessionReconciler.getNonTerminalJobs(resourceContext);
 
         assertEquals(
                 0,
-                unmanagedJobsAfterRemoval.size(),
-                "Should have no unmanaged jobs when only managed and terminated jobs exist");
+                nonTerminalJobsAfterRemoval.size(),
+                "Should have no non-terminal jobs when only terminated jobs exist");
     }
 }