Adds check for exceptions when prevState is terminal

Author: vsantwana

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java

@@ -41,7 +41,6 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
-import java.util.regex.Pattern;
 
 import static org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.updateFlinkResourceException;
 
@@ -51,8 +50,6 @@ public class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> {
     private static final Logger LOG = LoggerFactory.getLogger(JobStatusObserver.class);
 
     public static final String JOB_NOT_FOUND_ERR = "Job Not Found";
-    private static final Pattern VALID_K8S_ANNOTATION_KEY_PATTERN =
-            Pattern.compile("^[a-zA-Z0-9./-]{1,63}$");
 
     protected final EventRecorder eventRecorder;
 
@@ -90,7 +87,6 @@ public boolean observe(FlinkResourceContext<R> ctx) {
                 updateJobStatus(ctx, newJobStatus);
                 ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
                 // see if the JM server is up, try to get the exceptions
-                // in case the new
                 if (!previousJobStatus.isGloballyTerminalState()) {
                     observeJobManagerExceptions(ctx);
                 }

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java

@@ -157,6 +157,85 @@ void testFailed() throws Exception {
         assertTrue(flinkResourceEventCollector.events.isEmpty());
     }
 
+    @Test
+    public void testExceptionObservedEvenWhenNewStateIsTerminal() throws Exception {
+        var deployment = initDeployment();
+        var status = deployment.getStatus();
+        var jobStatus = status.getJobStatus();
+        jobStatus.setState(JobStatus.RUNNING);
+        Map<String, String> configuration = new HashMap<>();
+        configuration.put(
+                KubernetesOperatorConfigOptions.OPERATOR_EVENT_EXCEPTION_LIMIT.key(), "2");
+        Configuration operatorConfig = Configuration.fromMap(configuration);
+        FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
+                getResourceContext(deployment, operatorConfig);
+
+        var jobId = JobID.fromHexString(deployment.getStatus().getJobStatus().getJobId());
+        ctx.getExceptionCacheEntry().setJobId(jobId.toHexString());
+        ctx.getExceptionCacheEntry().setLastTimestamp(500L);
+        flinkService.addExceptionHistory(jobId, "ExceptionOne", "trace1", 1000L);
+
+        // Ensure jobFailedErr is null before the observe call
+        flinkService.submitApplicationCluster(
+                deployment.getSpec().getJob(), ctx.getDeployConfig(deployment.getSpec()), false);
+        flinkService.cancelJob(JobID.fromHexString(jobStatus.getJobId()), false);
+        flinkService.setJobFailedErr(null);
+
+        observer.observe(ctx);
+
+        var events =
+                kubernetesClient
+                        .v1()
+                        .events()
+                        .inNamespace(deployment.getMetadata().getNamespace())
+                        .list()
+                        .getItems();
+        assertEquals(2, events.size()); // one will be for job status changed
+        // assert that none of the events contain JOB_NOT_FOUND_ERR
+        assertFalse(
+                events.stream()
+                        .anyMatch(
+                                event ->
+                                        event.getMessage()
+                                                .contains(JobStatusObserver.JOB_NOT_FOUND_ERR)));
+    }
+
+    @Test
+    public void testExceptionNotObservedWhenOldStateIsTerminal() throws Exception {
+        var deployment = initDeployment();
+        var status = deployment.getStatus();
+        var jobStatus = status.getJobStatus();
+        jobStatus.setState(JobStatus.CANCELED);
+        Map<String, String> configuration = new HashMap<>();
+        configuration.put(
+                KubernetesOperatorConfigOptions.OPERATOR_EVENT_EXCEPTION_LIMIT.key(), "2");
+        Configuration operatorConfig = Configuration.fromMap(configuration);
+        FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
+                getResourceContext(deployment, operatorConfig);
+
+        var jobId = JobID.fromHexString(deployment.getStatus().getJobStatus().getJobId());
+        ctx.getExceptionCacheEntry().setJobId(jobId.toHexString());
+        ctx.getExceptionCacheEntry().setLastTimestamp(500L);
+        flinkService.addExceptionHistory(jobId, "ExceptionOne", "trace1", 1000L);
+
+        // Ensure jobFailedErr is null before the observe call
+        flinkService.submitApplicationCluster(
+                deployment.getSpec().getJob(), ctx.getDeployConfig(deployment.getSpec()), false);
+        flinkService.setJobFailedErr(null);
+
+        observer.observe(ctx);
+
+        var events =
+                kubernetesClient
+                        .v1()
+                        .events()
+                        .inNamespace(deployment.getMetadata().getNamespace())
+                        .list()
+                        .getItems();
+        assertEquals(1, events.size()); // only one event for job status changed
+        assertEquals(EventRecorder.Reason.JobStatusChanged.name(), events.get(0).getReason());
+    }
+
     @Test
     public void testExceptionLimitConfig() throws Exception {
         var observer = new JobStatusObserver<>(eventRecorder);