JobWatcher should not react to responses that are not for the current job being watched

Author: alai8

operator/src/main/java/oracle/kubernetes/operator/JobWatcher.java

@@ -83,10 +83,11 @@ public void receivedResponse(Watch.Response<V1Job> item) {
       case "ADDED":
       case "MODIFIED":
         V1Job job = item.object;
-        Boolean isComplete = isComplete(job) || isFailed(job);
+        Boolean isComplete = isComplete(job);
+        Boolean isFailed = isFailed(job);
         String jobName = job.getMetadata().getName();
-        if (isComplete) {
-          Complete complete = completeCallbackRegistrations.remove(jobName);
+        if (isComplete || isFailed) {
+          Complete complete = completeCallbackRegistrations.get(jobName);
           if (complete != null) {
             complete.isComplete(job);
           }
@@ -150,6 +151,11 @@ private WaitForJobReadyStep(V1Job job, Step next) {
       this.job = job;
     }
 
+    boolean shouldProcessJob(V1Job job) {
+      return (this.job.getMetadata().getCreationTimestamp().getMillis()
+          == job.getMetadata().getCreationTimestamp().getMillis());
+    }
+
     @Override
     public NextAction apply(Packet packet) {
       if (isComplete(job)) {
@@ -165,6 +171,10 @@ public NextAction apply(Packet packet) {
           (fiber) -> {
             Complete complete =
                 (V1Job job) -> {
+                  if (!shouldProcessJob(job)) {
+                    return;
+                  }
+                  completeCallbackRegistrations.remove(job.getMetadata().getName());
                   if (didResume.compareAndSet(false, true)) {
                     LOGGER.fine("Job status: " + job.getStatus());
                     packet.put(ProcessingConstants.DOMAIN_INTROSPECTOR_JOB, job);