Ensure that restart delay is set before any other activity is executed (

airbytehq#18547) * Ensure that restart delay is set before any other activity is executed * Add version check * Formatting
jhammarstedt · Oct 31, 2022 · f6371f0 · f6371f0
1 parent 097f51a
commit f6371f0
Showing 1 changed file with 10 additions and 2 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
@@ -121,6 +121,8 @@ public class ConnectionManagerWorkflowImpl implements ConnectionManagerWorkflow
   private static final int RECORD_METRIC_CURRENT_VERSION = 1;
   private static final String WORKFLOW_CONFIG_TAG = "workflow_config";
   private static final int WORKFLOW_CONFIG_CURRENT_VERSION = 1;
+  private static final String WORKFLOW_DELAY_TAG = "get_workflow_delay";
+  private static final int WORKFLOW_DELAY_TAG_CURRENT_VERSION = 1;
 
   private static final String ROUTE_ACTIVITY_TAG = "route_activity";
   private static final int ROUTE_ACTIVITY_CURRENT_VERSION = 1;
@@ -161,8 +163,14 @@ public class ConnectionManagerWorkflowImpl implements ConnectionManagerWorkflow
   public void run(final ConnectionUpdaterInput connectionUpdaterInput) throws RetryableException {
     try {
       ApmTraceUtils.addTagsToTrace(Map.of(CONNECTION_ID_KEY, connectionUpdaterInput.getConnectionId()));
-      recordMetric(new RecordMetricInput(connectionUpdaterInput, Optional.empty(), OssMetricsRegistry.TEMPORAL_WORKFLOW_ATTEMPT, null));
-      workflowDelay = getWorkflowRestartDelaySeconds();
+      if (Workflow.getVersion(WORKFLOW_DELAY_TAG, Workflow.DEFAULT_VERSION,
+          WORKFLOW_DELAY_TAG_CURRENT_VERSION) < WORKFLOW_DELAY_TAG_CURRENT_VERSION) {
+        recordMetric(new RecordMetricInput(connectionUpdaterInput, Optional.empty(), OssMetricsRegistry.TEMPORAL_WORKFLOW_ATTEMPT, null));
+        workflowDelay = getWorkflowRestartDelaySeconds();
+      } else {
+        workflowDelay = getWorkflowRestartDelaySeconds();
+        recordMetric(new RecordMetricInput(connectionUpdaterInput, Optional.empty(), OssMetricsRegistry.TEMPORAL_WORKFLOW_ATTEMPT, null));
+      }
 
       try {
         cancellableSyncWorkflow = generateSyncWorkflowRunnable(connectionUpdaterInput);