More comprehensive temporal error message (#18608)

* To be remove

* Remove the signal for waiting after a failed activity and ensure we are waiting the expected time

* Revert "To be remove"

This reverts commit 3a5f7b4.

* Remove unused and move failure reason to the helper

* Avoid repetitive new Set()

Author: benmoriceau

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

@@ -62,13 +62,6 @@ public interface ConnectionManagerWorkflow {
   @SignalMethod
   void resetConnectionAndSkipNextScheduling();
 
-  /**
-   * If an activity fails the workflow will be stuck. This signal activity can be used to retry the
-   * activity.
-   */
-  @SignalMethod
-  void retryFailedActivity();
-
   /**
    * Return the current state of the workflow.
    */

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

@@ -32,7 +32,6 @@ public WorkflowState(final UUID id, final WorkflowStateChangedListener stateChan
   private final boolean resetConnection = false;
   @Deprecated
   private final boolean continueAsReset = false;
-  private boolean retryFailedActivity = false;
   private boolean quarantined = false;
   private boolean success = true;
   private boolean cancelledForReset = false;
@@ -89,14 +88,6 @@ public void setFailed(final boolean failed) {
     this.failed = failed;
   }
 
-  public void setRetryFailedActivity(final boolean retryFailedActivity) {
-    final ChangedStateEvent event = new ChangedStateEvent(
-        StateField.RETRY_FAILED_ACTIVITY,
-        retryFailedActivity);
-    stateChangedListener.addEvent(id, event);
-    this.retryFailedActivity = retryFailedActivity;
-  }
-
   public void setQuarantined(final boolean quarantined) {
     final ChangedStateEvent event = new ChangedStateEvent(
         StateField.QUARANTINED,
@@ -146,7 +137,6 @@ public void reset() {
     this.setUpdated(false);
     this.setCancelled(false);
     this.setFailed(false);
-    this.setRetryFailedActivity(false);
     this.setSuccess(false);
     this.setQuarantined(false);
     this.setDoneWaiting(false);

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

@@ -31,7 +31,6 @@ enum StateField {
     FAILED,
     RESET,
     CONTINUE_AS_RESET,
-    RETRY_FAILED_ACTIVITY,
     QUARANTINED,
     SUCCESS,
     CANCELLED_FOR_RESET,

airbyte-commons-worker/src/main/java/io/airbyte/workers/helper/FailureHelper.java

@@ -227,6 +227,12 @@ public static FailureReason failureReasonFromWorkflowAndActivity(
     }
   }
 
+  public static FailureReason platformFailure(final Throwable t, final Long jobId, final Integer attemptNumber) {
+    return genericFailure(t, jobId, attemptNumber)
+        .withFailureOrigin(FailureOrigin.AIRBYTE_PLATFORM)
+        .withExternalMessage("Something went wrong within the airbyte platform");
+  }
+
   private static Metadata jobAndAttemptMetadata(final Long jobId, final Integer attemptNumber) {
     return new Metadata()
         .withAdditionalProperty(JOB_ID_METADATA_KEY, jobId)

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

@@ -85,6 +85,7 @@
 import io.temporal.workflow.Workflow;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -327,23 +328,31 @@ private void reportSuccess(final ConnectionUpdaterInput connectionUpdaterInput,
   private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput,
                              final StandardSyncOutput standardSyncOutput,
                              final FailureCause failureCause) {
+    reportFailure(connectionUpdaterInput, standardSyncOutput, failureCause, new HashSet<>());
+  }
+
+  private void reportFailure(final ConnectionUpdaterInput connectionUpdaterInput,
+                             final StandardSyncOutput standardSyncOutput,
+                             final FailureCause failureCause,
+                             final Set<FailureReason> failureReasonsOverride) {
     final int attemptCreationVersion =
         Workflow.getVersion(RENAME_ATTEMPT_ID_TO_NUMBER_TAG, Workflow.DEFAULT_VERSION, RENAME_ATTEMPT_ID_TO_NUMBER_CURRENT_VERSION);
 
+    final Set<FailureReason> failureReasons = failureReasonsOverride.isEmpty() ? workflowInternalState.getFailures() : failureReasonsOverride;
     if (attemptCreationVersion < RENAME_ATTEMPT_ID_TO_NUMBER_CURRENT_VERSION) {
       runMandatoryActivity(jobCreationAndStatusUpdateActivity::attemptFailure, new AttemptFailureInput(
           workflowInternalState.getJobId(),
           workflowInternalState.getAttemptNumber(),
           connectionUpdaterInput.getConnectionId(),
           standardSyncOutput,
-          FailureHelper.failureSummary(workflowInternalState.getFailures(), workflowInternalState.getPartialSuccess())));
+          FailureHelper.failureSummary(failureReasons, workflowInternalState.getPartialSuccess())));
     } else {
       runMandatoryActivity(jobCreationAndStatusUpdateActivity::attemptFailureWithAttemptNumber, new AttemptNumberFailureInput(
           workflowInternalState.getJobId(),
           workflowInternalState.getAttemptNumber(),
           connectionUpdaterInput.getConnectionId(),
           standardSyncOutput,
-          FailureHelper.failureSummary(workflowInternalState.getFailures(), workflowInternalState.getPartialSuccess())));
+          FailureHelper.failureSummary(failureReasons, workflowInternalState.getPartialSuccess())));
     }
 
     final int maxAttempt = configFetchActivity.getMaxAttempt().getMaxAttempt();
@@ -539,13 +548,6 @@ public void resetConnectionAndSkipNextScheduling() {
     }
   }
 
-  @Trace(operationName = WORKFLOW_TRACE_OPERATION_NAME)
-  @Override
-  public void retryFailedActivity() {
-    traceConnectionId();
-    workflowState.setRetryFailedActivity(true);
-  }
-
   @Trace(operationName = WORKFLOW_TRACE_OPERATION_NAME)
   @Override
   public WorkflowState getState() {
@@ -625,13 +627,14 @@ private <INPUT, OUTPUT> OUTPUT runMandatoryActivityWithOutput(final Function<INP
       // overwhelming temporal.
       log.info("Waiting {} before restarting the workflow for connection {}, to prevent spamming temporal with restarts.", workflowDelay,
           connectionId);
-      Workflow.await(workflowDelay, () -> workflowState.isRetryFailedActivity());
-
-      // Accept a manual signal to retry the failed activity during this window
-      if (workflowState.isRetryFailedActivity()) {
-        log.info("Received RetryFailedActivity signal for connection {}. Retrying activity.", connectionId);
-        workflowState.setRetryFailedActivity(false);
-        return runMandatoryActivityWithOutput(mapper, input);
+      Workflow.sleep(workflowDelay);
+
+      // If a jobId exist set the failure reason
+      if (workflowInternalState.getJobId() != null) {
+        final ConnectionUpdaterInput connectionUpdaterInput = connectionUpdaterInputFromState();
+        final FailureReason failureReason =
+            FailureHelper.platformFailure(e, workflowInternalState.getJobId(), workflowInternalState.getAttemptNumber());
+        reportFailure(connectionUpdaterInput, null, FailureCause.ACTIVITY, Set.of(failureReason));
       }
 
       log.info("Finished wait for connection {}, restarting connection manager workflow", connectionId);
@@ -645,6 +648,16 @@ private <INPUT, OUTPUT> OUTPUT runMandatoryActivityWithOutput(final Function<INP
     }
   }
 
+  private ConnectionUpdaterInput connectionUpdaterInputFromState() {
+    return ConnectionUpdaterInput.builder()
+        .connectionId(connectionId)
+        .jobId(workflowInternalState.getJobId())
+        .attemptNumber(workflowInternalState.getAttemptNumber())
+        .fromFailure(false)
+        .build();
+
+  }
+
   /**
    * Similar to runMandatoryActivityWithOutput but for methods that don't return
    */

airbyte-workers/src/test/java/io/airbyte/workers/temporal/scheduling/ConnectionManagerWorkflowTest.java

@@ -264,6 +264,7 @@ void runSuccess() throws InterruptedException {
       returnTrueForLastJobOrAttemptFailure();
       when(mConfigFetchActivity.getTimeToWait(Mockito.any()))
           .thenReturn(new ScheduleRetrieverOutput(SCHEDULE_WAIT));
+      when(mConfigFetchActivity.getMaxAttempt()).thenReturn(new GetMaxAttemptOutput(1));
 
       final UUID testId = UUID.randomUUID();
       final TestStateListener testStateListener = new TestStateListener();
@@ -308,6 +309,7 @@ void retryAfterFail() throws InterruptedException {
       returnTrueForLastJobOrAttemptFailure();
       when(mConfigFetchActivity.getTimeToWait(Mockito.any()))
           .thenReturn(new ScheduleRetrieverOutput(SCHEDULE_WAIT));
+      when(mConfigFetchActivity.getMaxAttempt()).thenReturn(new GetMaxAttemptOutput(1));
 
       final UUID testId = UUID.randomUUID();
       final TestStateListener testStateListener = new TestStateListener();
@@ -1475,6 +1477,7 @@ void testWorkflowRestartedAfterFailedActivity(final Thread mockSetup) throws Int
       mockSetup.run();
       when(mConfigFetchActivity.getTimeToWait(Mockito.any())).thenReturn(new ScheduleRetrieverOutput(
           Duration.ZERO));
+      when(mConfigFetchActivity.getMaxAttempt()).thenReturn(new GetMaxAttemptOutput(1));
 
       final UUID testId = UUID.randomUUID();
       TestStateListener.reset();
@@ -1505,42 +1508,6 @@ void testWorkflowRestartedAfterFailedActivity(final Thread mockSetup) throws Int
       assertWorkflowWasContinuedAsNew();
     }
 
-    @Test
-    void testCanRetryFailedActivity() throws InterruptedException {
-      returnTrueForLastJobOrAttemptFailure();
-      when(mJobCreationAndStatusUpdateActivity.createNewJob(Mockito.any()))
-          .thenThrow(ApplicationFailure.newNonRetryableFailure("", ""))
-          .thenReturn(new JobCreationOutput(1l));
-
-      when(mConfigFetchActivity.getTimeToWait(Mockito.any())).thenReturn(new ScheduleRetrieverOutput(
-          Duration.ZERO));
-
-      final UUID testId = UUID.randomUUID();
-      final TestStateListener testStateListener = new TestStateListener();
-      final WorkflowState workflowState = new WorkflowState(testId, testStateListener);
-
-      final ConnectionUpdaterInput input = ConnectionUpdaterInput.builder()
-          .connectionId(UUID.randomUUID())
-          .jobId(null)
-          .attemptId(null)
-          .fromFailure(false)
-          .attemptNumber(1)
-          .workflowState(workflowState)
-          .build();
-
-      startWorkflowAndWaitUntilReady(workflow, input);
-
-      // Sleep test env for half of restart delay, so that we know we are in the middle of the delay
-      testEnv.sleep(WORKFLOW_FAILURE_RESTART_DELAY.dividedBy(2));
-      workflow.retryFailedActivity();
-      Thread.sleep(500); // any time after no-waiting manual run
-      final Queue<ChangedStateEvent> events = testStateListener.events(testId);
-
-      Assertions.assertThat(events)
-          .filteredOn(changedStateEvent -> changedStateEvent.getField() == StateField.RETRY_FAILED_ACTIVITY && changedStateEvent.isValue())
-          .hasSize(1);
-    }
-
   }
 
   private class HasFailureFromOrigin implements ArgumentMatcher<AttemptNumberFailureInput> {