Add test that check that we continue a reset as a reset if it failed (#10806)

This is adding tests to make sure that a reset is continued as a reset after an attempt or as a job when the maximum amount of attempt is reach.

It also fixes the workflow to continue as a reset in a new job if it fails more than the maximum number of attempt.

Open question:

- Is it what we want for the job (continue as a reset if the job failed)?
- Do we need to respect the schedule if the reset failed more than the maximum attempts?

Author: benmoriceau

airbyte-workers/src/main/java/io/airbyte/workers/temporal/TemporalClient.java

@@ -219,7 +219,17 @@ public void submitConnectionUpdaterAsync(final UUID connectionId) {
     final ConnectionManagerWorkflow connectionManagerWorkflow = getWorkflowOptionsWithWorkflowId(ConnectionManagerWorkflow.class,
         TemporalJobType.CONNECTION_UPDATER, getConnectionManagerName(connectionId));
     final BatchRequest signalRequest = client.newSignalWithStartRequest();
-    final ConnectionUpdaterInput input = new ConnectionUpdaterInput(connectionId, null, null, false, 1, null, false);
+    final ConnectionUpdaterInput input = ConnectionUpdaterInput.builder()
+        .connectionId(connectionId)
+        .jobId(null)
+        .attemptId(null)
+        .fromFailure(false)
+        .attemptNumber(1)
+        .workflowState(null)
+        .resetConnection(false)
+        .fromJobResetFailure(false)
+        .build();
+
     signalRequest.add(connectionManagerWorkflow::run, input);
 
     WorkflowClient.start(connectionManagerWorkflow::run, input);

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflow.java

@@ -63,12 +63,6 @@ public interface ConnectionManagerWorkflow {
   @SignalMethod
   void retryFailedActivity();
 
-  /**
-   * Use for testing in order to simulate a job failure.
-   */
-  @SignalMethod
-  void simulateFailure();
-
   /**
    * Return the current state of the workflow.
    */

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowImpl.java

@@ -122,6 +122,8 @@ private CancellationScope generateSyncWorkflowRunnable(final ConnectionUpdaterIn
       // resetConnection flag to the next run so that that run can execute the actual reset
       workflowState.setResetConnection(connectionUpdaterInput.isResetConnection());
 
+      workflowState.setResetWithScheduling(connectionUpdaterInput.isFromJobResetFailure());
+
       final Duration timeToWait = getTimeToWait(connectionUpdaterInput.getConnectionId());
 
       Workflow.await(timeToWait,
@@ -209,10 +211,11 @@ private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput,
     final int maxAttempt = configFetchActivity.getMaxAttempt().getMaxAttempt();
     final int attemptNumber = connectionUpdaterInput.getAttemptNumber();
 
+    if (workflowState.isResetConnection()) {
+      workflowState.setContinueAsReset(true);
+    }
+
     if (maxAttempt > attemptNumber) {
-      if (workflowState.isResetConnection()) {
-        workflowState.setContinueAsReset(true);
-      }
       // restart from failure
       connectionUpdaterInput.setAttemptNumber(attemptNumber + 1);
       connectionUpdaterInput.setFromFailure(true);
@@ -222,6 +225,9 @@ private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput,
           "Job failed after too many retries for connection " + connectionId));
 
       resetNewConnectionInput(connectionUpdaterInput);
+      if (workflowState.isResetConnection()) {
+        connectionUpdaterInput.setFromJobResetFailure(true);
+      }
     }
   }
 
@@ -265,6 +271,7 @@ public void connectionUpdated() {
   @Override
   public void resetConnection() {
     workflowState.setResetConnection(true);
+    workflowState.setResetWithScheduling(false);
     if (workflowState.isRunning()) {
       workflowState.setCancelledForReset(true);
       cancellableSyncWorkflow.cancel();
@@ -276,11 +283,6 @@ public void retryFailedActivity() {
     workflowState.setRetryFailedActivity(true);
   }
 
-  @Override
-  public void simulateFailure() {
-    workflowState.setFailed(true);
-  }
-
   @Override
   public WorkflowState getState() {
     return workflowState;
@@ -313,7 +315,8 @@ public QuarantinedInformation getQuarantinedInformation() {
    * delete
    */
   private Boolean skipScheduling() {
-    return workflowState.isSkipScheduling() || workflowState.isDeleted() || workflowState.isUpdated() || workflowState.isResetConnection();
+    return workflowState.isSkipScheduling() || workflowState.isDeleted() || workflowState.isUpdated() ||
+        (!workflowState.isResetWithScheduling() && workflowState.isResetConnection());
   }
 
   private void prepareForNextRunAndContinueAsNew(final ConnectionUpdaterInput connectionUpdaterInput) {
@@ -495,10 +498,7 @@ private boolean getFailStatus(final StandardSyncOutput standardSyncOutput) {
       return true;
     }
 
-    // For testing purpose we simulate a failure using a signal method to avoid having to do a static
-    // mock.
-    // We do override failure reason in this case.
-    return false || workflowState.isFailed();
+    return false;
   }
 
   /**

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/ConnectionUpdaterInput.java

@@ -8,13 +8,15 @@
 import java.util.UUID;
 import javax.annotation.Nullable;
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.NonNull;
 
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
+@Builder
 public class ConnectionUpdaterInput {
 
   @NonNull
@@ -36,5 +38,6 @@ public class ConnectionUpdaterInput {
   @Nullable
   private WorkflowState workflowState;
   private boolean resetConnection;
+  private boolean fromJobResetFailure = false;
 
 }

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/state/WorkflowState.java

@@ -34,6 +34,7 @@ public WorkflowState(final UUID id, final WorkflowStateChangedListener stateChan
   private boolean quarantined = false;
   private boolean success = true;
   private boolean cancelledForReset = false;
+  private boolean resetWithScheduling = false;
 
   public void setRunning(final boolean running) {
     final ChangedStateEvent event = new ChangedStateEvent(
@@ -131,6 +132,16 @@ public void setCancelledForReset(final boolean cancelledForReset) {
     this.cancelledForReset = cancelledForReset;
   }
 
+  public void setResetWithScheduling(final boolean resetWithScheduling) {
+    final ChangedStateEvent event = new ChangedStateEvent(
+        StateField.RESET_WITH_SCHEDULING,
+        resetWithScheduling);
+    stateChangedListener.addEvent(id, event);
+    this.resetWithScheduling = resetWithScheduling;
+  }
+
+  // TODO: bmoric -> This is noisy when inpecting the list of event, it should be just a single reset
+  // event.
   public void reset() {
     this.setRunning(false);
     this.setDeleted(false);

airbyte-workers/src/main/java/io/airbyte/workers/temporal/scheduling/state/listener/WorkflowStateChangedListener.java

@@ -35,6 +35,7 @@ enum StateField {
     QUARANTINED,
     SUCCESS,
     CANCELLED_FOR_RESET,
+    RESET_WITH_SCHEDULING,
   }
 
   @Value