🐛 fix issue where jobs retried forever (#2502)

airbytehq · Mar 17, 2021 · 30ddcd7 · 30ddcd7
1 parent 6c8d110
commit 30ddcd7
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 1 deletion.
diff --git a/...yte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/TemporalWorkerRunFactory.java b/...yte-scheduler/src/main/java/io/airbyte/scheduler/worker_run/TemporalWorkerRunFactory.java
@@ -43,6 +43,7 @@
 import io.airbyte.workers.temporal.TemporalClient;
 import io.airbyte.workers.temporal.TemporalJobException;
 import io.airbyte.workers.temporal.TemporalJobType;
+import io.temporal.failure.TemporalException;
 import java.nio.file.Path;
 
 public class TemporalWorkerRunFactory {
@@ -122,7 +123,11 @@ private static TemporalJobType toTemporalJobType(ConfigType jobType) {
   private OutputAndStatus<JobOutput> toOutputAndStatus(CheckedSupplier<JobOutput, TemporalJobException> supplier) {
     try {
       return new OutputAndStatus<>(JobStatus.SUCCEEDED, supplier.get());
-    } catch (TemporalJobException e) {
+    } catch (TemporalJobException | TemporalException e) {
+      // while from within the temporal activity we throw TemporalJobException, Temporal wraps any
+      // exception thrown by an activity in an ApplicationFailure exception, which gets wrapped in
+      // ActivityFailure exception which get wrapped in a WorkflowFailedException. We will need to unwrap
+      // these later, but for now we just catch the parent.
       return new OutputAndStatus<>(JobStatus.FAILED);
     }
   }

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/CheckConnectionWorkflow.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/CheckConnectionWorkflow.java
@@ -54,6 +54,7 @@ class WorkflowImpl implements CheckConnectionWorkflow {
 
     final ActivityOptions options = ActivityOptions.newBuilder()
         .setScheduleToCloseTimeout(Duration.ofHours(1))
+        .setRetryOptions(TemporalUtils.NO_RETRY)
         .build();
     private final CheckConnectionActivity activity = Workflow.newActivityStub(CheckConnectionActivity.class, options);
 

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/DiscoverCatalogWorkflow.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/DiscoverCatalogWorkflow.java
@@ -56,6 +56,7 @@ class WorkflowImpl implements DiscoverCatalogWorkflow {
 
     final ActivityOptions options = ActivityOptions.newBuilder()
         .setScheduleToCloseTimeout(Duration.ofHours(2))
+        .setRetryOptions(TemporalUtils.NO_RETRY)
         .build();
     private final DiscoverCatalogActivity activity = Workflow.newActivityStub(DiscoverCatalogActivity.class, options);
 

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/SpecWorkflow.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/SpecWorkflow.java
@@ -51,6 +51,7 @@ class WorkflowImpl implements SpecWorkflow {
 
     final ActivityOptions options = ActivityOptions.newBuilder()
         .setScheduleToCloseTimeout(Duration.ofHours(1))
+        .setRetryOptions(TemporalUtils.NO_RETRY)
         .build();
     private final SpecActivity activity = Workflow.newActivityStub(SpecActivity.class, options);
 

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/SyncWorkflow.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/SyncWorkflow.java
@@ -63,6 +63,7 @@ class WorkflowImpl implements SyncWorkflow {
 
     final ActivityOptions options = ActivityOptions.newBuilder()
         .setScheduleToCloseTimeout(Duration.ofDays(3))
+        .setRetryOptions(TemporalUtils.NO_RETRY)
         .build();
     private final SyncActivity activity = Workflow.newActivityStub(SyncActivity.class, options);
 

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalUtils.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalUtils.java
@@ -27,6 +27,7 @@
 import io.airbyte.scheduler.models.JobRunConfig;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
+import io.temporal.common.RetryOptions;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
 import java.io.Serializable;
@@ -43,6 +44,8 @@ public class TemporalUtils {
 
   public static final WorkflowClient TEMPORAL_CLIENT = WorkflowClient.newInstance(TEMPORAL_SERVICE);
 
+  public static final RetryOptions NO_RETRY = RetryOptions.newBuilder().setMaximumAttempts(1).build();
+
   @FunctionalInterface
   public interface TemporalJobCreator<T extends Serializable> {
 
@@ -52,6 +55,7 @@ public interface TemporalJobCreator<T extends Serializable> {
 
   public static WorkflowOptions getWorkflowOptions(TemporalJobType jobType) {
     return WorkflowOptions.newBuilder()
+        .setRetryOptions(NO_RETRY)
         .setTaskQueue(jobType.name())
         .build();
   }