Correct job watch processing

Author: rjeberhard

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

@@ -70,7 +70,7 @@ public class DomainNamespaces {
   private final WatcherControl<CoreV1Event, OperatorEventWatcher> operatorEventWatchers
       = new WatcherControl<>(OperatorEventWatcher::create, d -> d::dispatchEventWatch);
   private final WatcherControl<V1Job, JobWatcher> jobWatchers
-        = new WatcherControl<>(JobWatcher::create, d -> NULL_LISTENER);
+        = new WatcherControl<>(JobWatcher::create, d -> d::dispatchJobWatch);
   private final WatcherControl<V1Pod, PodWatcher> podWatchers
         = new WatcherControl<>(PodWatcher::create, d -> d::dispatchPodWatch);
   private final WatcherControl<V1Service, ServiceWatcher> serviceWatchers

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2018, 2023, Oracle and/or its affiliates.
+// Copyright (c) 2018, 2024, Oracle and/or its affiliates.
 // Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator;
@@ -10,6 +10,7 @@
 
 import io.kubernetes.client.openapi.models.CoreV1Event;
 import io.kubernetes.client.openapi.models.V1ConfigMap;
+import io.kubernetes.client.openapi.models.V1Job;
 import io.kubernetes.client.openapi.models.V1PersistentVolumeClaim;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodDisruptionBudget;
@@ -71,6 +72,12 @@ MakeRightClusterOperation createMakeRightOperationForClusterEvent(
    */
   void dispatchDomainWatch(Watch.Response<DomainResource> item);
 
+  /**
+   * Handles a watch event for jobs in the managed namespaces.
+   * @param item a Kubernetes watch event
+   */
+  void dispatchJobWatch(Watch.Response<V1Job> item);
+
   /**
    * Handles a watch event for pods in the managed namespaces.
    * @param item a Kubernetes watch event

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

@@ -19,6 +19,7 @@
 import io.kubernetes.client.extended.controller.reconciler.Result;
 import io.kubernetes.client.openapi.models.CoreV1Event;
 import io.kubernetes.client.openapi.models.V1ConfigMap;
+import io.kubernetes.client.openapi.models.V1Job;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.openapi.models.V1ObjectReference;
 import io.kubernetes.client.openapi.models.V1PersistentVolumeClaim;
@@ -51,6 +52,7 @@
 import oracle.kubernetes.operator.logging.ThreadLoggingContext;
 import oracle.kubernetes.operator.steps.BeforeAdminServiceStep;
 import oracle.kubernetes.operator.tuning.TuningParameters;
+import oracle.kubernetes.operator.watcher.JobWatcher;
 import oracle.kubernetes.operator.work.Cancellable;
 import oracle.kubernetes.operator.work.Fiber;
 import oracle.kubernetes.operator.work.Fiber.CompletionCallback;
@@ -543,6 +545,42 @@ private String getDomainUid(Fiber fiber) {
           .map(DomainPresenceInfo::getDomainUid).orElse("");
   }
 
+  /**
+   * Dispatch job watch event.
+   * @param item watch event
+   */
+  public void dispatchJobWatch(Watch.Response<V1Job> item) {
+    V1Job job = item.object;
+    String domainUid = getJobDomainUid(job);
+    String namespace = Optional.ofNullable(job.getMetadata()).map(V1ObjectMeta::getNamespace).orElse(null);
+    if (domainUid == null || namespace == null) {
+      return;
+    }
+
+    DomainPresenceInfo info = getExistingDomainPresenceInfo(namespace, domainUid);
+    if (info == null) {
+      return;
+    }
+
+    switch (item.type) {
+      case MODIFIED:
+        if (JobWatcher.isComplete(job) || JobWatcher.isFailed(job)) {
+          createMakeRightOperation(info).interrupt().withExplicitRecheck().execute();
+        }
+        break;
+      default:
+    }
+  }
+
+  private static String getJobDomainUid(V1Job job) {
+    return Optional.ofNullable(job)
+            .map(V1Job::getMetadata)
+            .map(V1ObjectMeta::getLabels)
+            .orElse(Collections.emptyMap())
+            .get(LabelConstants.DOMAINUID_LABEL);
+  }
+
+
   /**
    * Dispatch pod watch event.
    * @param item watch event

operator/src/main/java/oracle/kubernetes/operator/helpers/JobHelper.java

@@ -459,16 +459,16 @@ private Step checkForFailedIntrospectionJob() {
       return new CheckForFailedIntrospectorStep();
     }
 
-    private static class CheckForFailedIntrospectorStep extends Step {
+    private class CheckForFailedIntrospectorStep extends Step {
 
       @Override
       public @Nonnull Result apply(Packet packet) {
         V1Job domainIntrospectorJob = (V1Job) packet.get(ProcessingConstants.DOMAIN_INTROSPECTOR_JOB);
 
         if (JobWatcher.isFailed(domainIntrospectorJob)) {
           return doNext(
-              Step.chain(createIntrospectionFailureSteps(
-                  getFailedReason(domainIntrospectorJob), domainIntrospectorJob), getNext()), packet);
+              Step.chain(createIntrospectionFailureSteps(getFailedReason(domainIntrospectorJob), domainIntrospectorJob),
+                  cleanUpAndReintrospect(getNext())), packet);
         }
         if (JobWatcher.isComplete(domainIntrospectorJob)) {
           return doNext(createRemoveFailuresStep(getNext()), packet);