[GOBBLIN-2255] Temporal workflow fast failure

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/GobblinMultiTaskAttempt.java

@@ -348,13 +348,18 @@ private void persistTaskStateStore()
 
     if (hasTaskFailure) {
       String errorMsg = String.format("Tasks in container %s failed", containerIdOptional.or(""));
+      Throwable cause = null;
       for (Task task : tasks) {
         if (task.getTaskState().contains(ConfigurationKeys.TASK_FAILURE_EXCEPTION_KEY)) {
           errorMsg = String.format("Task failed: %s (Gobblin task id %s, container id %s)",
                                    task.getTaskState().getProp(ConfigurationKeys.TASK_FAILURE_EXCEPTION_KEY),
                                    task.getTaskId(), containerIdOptional.or(""));
         }
 
+        if (task.getTaskFailureException() != null) {
+          cause = task.getTaskFailureException();
+        }
+
         // If there are task failures then the tasks may be reattempted. Save a copy of the task state that is used
         // to filter out successful tasks on subsequent attempts.
         if (task.getTaskState().getWorkingState() == WorkUnitState.WorkingState.SUCCESSFUL
@@ -364,7 +369,7 @@ private void persistTaskStateStore()
         }
       }
 
-      throw new IOException(errorMsg);
+      throw new IOException(errorMsg, cause);
     }
   }
 

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/Task.java

@@ -136,6 +136,8 @@ public class Task implements TaskIFace {
   private static final String TASK_STATE = "taskState";
   private static final String FAILED_TASK_EVENT = "failedTask";
 
+  private volatile Throwable taskFailureException;
+
   private final String jobId;
   private final String taskId;
   private final String taskKey;
@@ -617,6 +619,7 @@ protected void onRecordExtract() {
 
   protected void failTask(Throwable t) {
     Throwable cleanedException = ExceptionUtils.removeEmptyWrappers(t);
+    this.taskFailureException = cleanedException;
 
     LOG.error(String.format("Task %s failed", this.taskId), cleanedException);
     this.taskState.setWorkingState(WorkUnitState.WorkingState.FAILED);
@@ -634,6 +637,10 @@ protected void failTask(Throwable t) {
     }
   }
 
+  public Throwable getTaskFailureException() {
+    return taskFailureException;
+  }
+
   /**
    * Whether the task should directly publish its output data to the final publisher output directory.
    *

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/ProcessWorkUnitImpl.java

@@ -33,6 +33,7 @@
 import com.google.common.collect.Lists;
 import io.temporal.activity.Activity;
 import io.temporal.activity.ActivityExecutionContext;
+import io.temporal.failure.ApplicationFailure;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -59,6 +60,7 @@
 import org.apache.gobblin.source.workunit.WorkUnit;
 import org.apache.gobblin.temporal.ddm.activity.ProcessWorkUnit;
 import org.apache.gobblin.temporal.ddm.util.JobStateUtils;
+import org.apache.gobblin.temporal.ddm.util.NonRetryableExceptions;
 import org.apache.gobblin.temporal.ddm.work.WorkUnitClaimCheck;
 import org.apache.gobblin.temporal.ddm.work.assistance.Help;
 import org.apache.gobblin.util.ExecutorsUtils;
@@ -91,7 +93,17 @@ public int processWorkUnit(WorkUnitClaimCheck wu) {
       troubleshooter = AutomaticTroubleshooterFactory.createForJob(jobState.getProperties());
       troubleshooter.start();
       return execute(workUnits, wu, jobState, fs, troubleshooter.getIssueRepository(), jobState.getProperties());
-    } catch (IOException | InterruptedException e) {
+    } catch (IOException e) {
+      Optional<Throwable> nonRetryable = NonRetryableExceptions.matchNonRetryable(e);
+      if (nonRetryable.isPresent()) {
+        Throwable cause = nonRetryable.get();
+        String errMsg = String.format("%s - non-retryable failure (%s): %s",
+            correlator, cause.getClass().getSimpleName(), cause.getMessage());
+        log.error(errMsg, e);
+        throw ApplicationFailure.newNonRetryableFailure(errMsg, cause.getClass().getSimpleName(), cause);
+      }
+      throw new RuntimeException(e);
+    } catch (InterruptedException e) {
       throw new RuntimeException(e);
     } finally {
       Help.finalizeTroubleshooting(troubleshooter, eventSubmitter, log, correlator);

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/NonRetryableExceptions.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
+import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
+
+import lombok.experimental.UtilityClass;
+
+
+/**
+ * Defines exception types that should trigger immediate, non-retryable application failure
+ * in Temporal workflows. These represent unrecoverable conditions where retrying would be futile.
+ */
+@UtilityClass
+public class NonRetryableExceptions {
+
+  private static final Set<Class<? extends Exception>> EXCEPTIONS;
+  static {
+    Set<Class<? extends Exception>> exceptions = new HashSet<>();
+    exceptions.add(NSQuotaExceededException.class);
+    exceptions.add(DSQuotaExceededException.class);
+    EXCEPTIONS = Collections.unmodifiableSet(exceptions);
+  }
+
+  /**
+   * Checks whether the given throwable has a cause that matches a known non-retryable exception type.
+   * @return an {@link Optional} containing the matched throwable if it is non-retryable, empty otherwise.
+   */
+  public static Optional<Throwable> matchNonRetryable(Throwable throwable) {
+    while (throwable != null) {
+      for (Class<? extends Exception> exClass : EXCEPTIONS) {
+        if (exClass.isInstance(throwable)) {
+          return Optional.of(throwable);
+        }
+      }
+      throwable = throwable.getCause();
+    }
+
+    return Optional.empty();
+  }
+}

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/util/NonRetryableExceptionsTest.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gobblin.temporal.ddm.util;
+
+import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
+import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+
+
+public class NonRetryableExceptionsTest {
+
+  @Test
+  public void testMatchesQuotaExceededExceptions() {
+    NSQuotaExceededException nsEx = new NSQuotaExceededException("ns quota exceeded");
+    Assert.assertTrue(NonRetryableExceptions.matchNonRetryable(nsEx).isPresent());
+    Assert.assertSame(NonRetryableExceptions.matchNonRetryable(nsEx).get(), nsEx);
+
+    DSQuotaExceededException dsEx = new DSQuotaExceededException("ds quota exceeded");
+    Assert.assertTrue(NonRetryableExceptions.matchNonRetryable(dsEx).isPresent());
+    Assert.assertSame(NonRetryableExceptions.matchNonRetryable(dsEx).get(), dsEx);
+
+    IOException ioEx = new IOException(dsEx);
+    Assert.assertTrue(NonRetryableExceptions.matchNonRetryable(ioEx).isPresent());
+    Assert.assertSame(NonRetryableExceptions.matchNonRetryable(ioEx).get(), dsEx);
+
+    IOException ioEx2 = new IOException(ioEx);
+    Assert.assertTrue(NonRetryableExceptions.matchNonRetryable(ioEx2).isPresent());
+    Assert.assertSame(NonRetryableExceptions.matchNonRetryable(ioEx2).get(), dsEx);
+
+    IOException ioEx3 = new IOException(ioEx2);
+    Assert.assertTrue(NonRetryableExceptions.matchNonRetryable(ioEx3).isPresent());
+    Assert.assertSame(NonRetryableExceptions.matchNonRetryable(ioEx3).get(), dsEx);
+  }
+
+  @Test
+  public void testDoesNotMatchArbitraryException() {
+    Assert.assertFalse(NonRetryableExceptions.matchNonRetryable(new IOException("some error")).isPresent());
+
+    Assert.assertFalse(NonRetryableExceptions.matchNonRetryable(new IOException(new IOException("some error"))).isPresent());
+  }
+
+  @Test
+  public void testReturnsEmptyForNull() {
+    Assert.assertFalse(NonRetryableExceptions.matchNonRetryable(null).isPresent());
+  }
+}