Ignore auto-disable activity if connection already inactive (#12507)

* Ignore auto disable activity if already inactive * fix formatting * fix unit test
airbytehq · May 2, 2022 · 4ab93e5 · 4ab93e5
1 parent de70351
commit 4ab93e5
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 7 deletions.
diff --git a/.../io/airbyte/workers/temporal/scheduling/activities/AutoDisableConnectionActivityImpl.java b/.../io/airbyte/workers/temporal/scheduling/activities/AutoDisableConnectionActivityImpl.java
@@ -13,7 +13,6 @@
 import io.airbyte.config.Configs;
 import io.airbyte.config.StandardSync;
 import io.airbyte.config.StandardSync.Status;
-import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.scheduler.models.Job;
 import io.airbyte.scheduler.models.JobStatus;
@@ -25,7 +24,6 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Optional;
-import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import lombok.AllArgsConstructor;
 
@@ -50,6 +48,12 @@ public class AutoDisableConnectionActivityImpl implements AutoDisableConnectionA
   public AutoDisableConnectionOutput autoDisableFailingConnection(final AutoDisableConnectionActivityInput input) {
     if (featureFlags.autoDisablesFailingConnections()) {
       try {
+        // if connection is already inactive, no need to disable
+        final StandardSync standardSync = configRepository.getStandardSync(input.getConnectionId());
+        if (standardSync.getStatus() == Status.INACTIVE) {
+          return new AutoDisableConnectionOutput(false);
+        }
+
         final int maxDaysOfOnlyFailedJobs = configs.getMaxDaysOfOnlyFailedJobsBeforeConnectionDisable();
         final int maxDaysOfOnlyFailedJobsBeforeWarning = maxDaysOfOnlyFailedJobs / 2;
         final int maxFailedJobsInARowBeforeConnectionDisableWarning = configs.getMaxFailedJobsInARowBeforeConnectionDisable() / 2;
@@ -82,7 +86,7 @@ public AutoDisableConnectionOutput autoDisableFailingConnection(final AutoDisabl
           return new AutoDisableConnectionOutput(false);
         } else if (numFailures >= configs.getMaxFailedJobsInARowBeforeConnectionDisable()) {
           // disable connection if max consecutive failed jobs limit has been hit
-          disableConnection(input.getConnectionId(), lastJob);
+          disableConnection(standardSync, lastJob);
           return new AutoDisableConnectionOutput(true);
         } else if (numFailures == maxFailedJobsInARowBeforeConnectionDisableWarning && !warningPreviouslySentForMaxDays) {
           // warn if number of consecutive failures hits 50% of MaxFailedJobsInARow
@@ -102,7 +106,7 @@ public AutoDisableConnectionOutput autoDisableFailingConnection(final AutoDisabl
 
         // disable connection if only failed jobs in the past maxDaysOfOnlyFailedJobs days
         if (firstReplicationOlderThanMaxDisableDays && noPreviousSuccess) {
-          disableConnection(input.getConnectionId(), lastJob);
+          disableConnection(standardSync, lastJob);
           return new AutoDisableConnectionOutput(true);
         }
 
@@ -172,8 +176,7 @@ private int getDaysSinceTimestamp(final long currentTimestampInSeconds, final lo
     return Math.toIntExact(TimeUnit.SECONDS.toDays(currentTimestampInSeconds - timestampInSeconds));
   }
 
-  private void disableConnection(final UUID connectionId, final Job lastJob) throws JsonValidationException, IOException, ConfigNotFoundException {
-    final StandardSync standardSync = configRepository.getStandardSync(connectionId);
+  private void disableConnection(final StandardSync standardSync, final Job lastJob) throws JsonValidationException, IOException {
     standardSync.setStatus(Status.INACTIVE);
     configRepository.writeStandardSync(standardSync);
 

diff --git a/.../io/airbyte/workers/temporal/scheduling/activities/AutoDisableConnectionActivityTest.java b/.../io/airbyte/workers/temporal/scheduling/activities/AutoDisableConnectionActivityTest.java
@@ -82,7 +82,8 @@ class AutoDisableConnectionActivityTest {
   private final StandardSync standardSync = new StandardSync();
 
   @BeforeEach
-  void setUp() throws IOException {
+  void setUp() throws IOException, JsonValidationException, ConfigNotFoundException {
+    Mockito.when(mConfigRepository.getStandardSync(CONNECTION_ID)).thenReturn(standardSync);
     standardSync.setStatus(Status.ACTIVE);
     Mockito.when(mFeatureFlags.autoDisablesFailingConnections()).thenReturn(true);
     Mockito.when(mConfigs.getMaxDaysOfOnlyFailedJobsBeforeConnectionDisable()).thenReturn(MAX_DAYS_OF_ONLY_FAILED_JOBS);