Update the cancellation (airbytehq#9705)

frankfanslc · Jan 21, 2022 · 19ea1b6 · 19ea1b6
1 parent 0864c00
commit 19ea1b6
Showing 1 changed file with 9 additions and 6 deletions.
diff --git a/...s/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java b/...s/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java
@@ -167,6 +167,7 @@ public void run(final ConnectionUpdaterInput connectionUpdaterInput) throws Retr
       } else if (workflowState.isCancelled()) {
         jobCreationAndStatusUpdateActivity.jobCancelled(new JobCancelledInput(
             maybeJobId.get()));
+        resetNewConnectionInput(connectionUpdaterInput);
       } else if (workflowState.isFailed()) {
         reportFailure(connectionUpdaterInput);
       } else {
@@ -188,9 +189,7 @@ private void reportSuccess(final ConnectionUpdaterInput connectionUpdaterInput)
         maybeAttemptId.get(),
         standardSyncOutput.orElse(null)));
 
-    connectionUpdaterInput.setJobId(null);
-    connectionUpdaterInput.setAttemptNumber(1);
-    connectionUpdaterInput.setFromFailure(false);
+    resetNewConnectionInput(connectionUpdaterInput);
   }
 
   private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput) {
@@ -212,12 +211,16 @@ private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput)
 
       Workflow.await(Duration.ofMinutes(1), () -> skipScheduling());
 
-      connectionUpdaterInput.setJobId(null);
-      connectionUpdaterInput.setAttemptNumber(1);
-      connectionUpdaterInput.setFromFailure(false);
+      resetNewConnectionInput(connectionUpdaterInput);
     }
   }
 
+  private void resetNewConnectionInput(ConnectionUpdaterInput connectionUpdaterInput) {
+    connectionUpdaterInput.setJobId(null);
+    connectionUpdaterInput.setAttemptNumber(1);
+    connectionUpdaterInput.setFromFailure(false);
+  }
+
   @Override
   public void submitManualSync() {
     if (workflowState.isRunning()) {