skip rest of introspector steps if job failed with DeadlineExceeded

Author: alai8

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

@@ -129,6 +129,18 @@ public static boolean isFailed(V1Job job) {
     return false;
   }
 
+  public static String getFailedReason(V1Job job) {
+    V1JobStatus status = job.getStatus();
+    if (status != null && status.getConditions() != null) {
+      for (V1JobCondition cond : status.getConditions()) {
+        if ("Failed".equals(cond.getType()) && "True".equals(cond.getStatus())) {
+          return cond.getReason();
+        }
+      }
+    }
+    return null;
+  }
+
   @Override
   public WatchI<V1Job> initiateWatch(WatchBuilder watchBuilder) throws ApiException {
     return watchBuilder
@@ -150,7 +162,7 @@ public void receivedResponse(Watch.Response<V1Job> item) {
         if (isComplete || isFailed) {
           Complete complete = completeCallbackRegistrations.get(jobName);
           if (complete != null) {
-            complete.isComplete(job);
+            complete.isComplete(job, isFailed);
           }
         }
         break;
@@ -175,7 +187,7 @@ public Step waitForReady(V1Job job, Step next) {
 
   @FunctionalInterface
   private interface Complete {
-    void isComplete(V1Job job);
+    void isComplete(V1Job job, boolean isJobFailed);
   }
 
   static class JobWatcherFactory {
@@ -231,14 +243,21 @@ public NextAction apply(Packet packet) {
       return doSuspend(
           (fiber) -> {
             Complete complete =
-                (V1Job job) -> {
+                (V1Job job, boolean isJobFailed) -> {
                   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);
+                    // Do not proceed to next step such as ReadDomainIntrospectorPodLog if job
+                    // failed due to DeadlineExceeded, as the pod container would likely not
+                    // be available for reading
+                    if (isJobFailed && "DeadlineExceeded".equals(getFailedReason(job))) {
+                      fiber.terminate(
+                          new DeadlineExceededException(job.getMetadata().getName()), packet);
+                    }
                     fiber.resume(packet);
                   }
                 };
@@ -283,4 +302,17 @@ public NextAction onSuccess(
           });
     }
   }
+
+  static class DeadlineExceededException extends Exception {
+    final String job;
+
+    public DeadlineExceededException(String job) {
+      super();
+      this.job = job;
+    }
+
+    public String toString() {
+      return "Job " + job + " failed. Reason: DeadlineExceeded";
+    }
+  }
 }