fix-mysql-cdc: poll for 5 minutes only when we have not received a si…

…ngle record (#3789) * fix-mysql-cdc: poll for 5 minutes only when we have not received a single record * fix format * address review comment + bump docker version
airbytehq · Jun 2, 2021 · cbf47ad · cbf47ad
1 parent bc44cae
commit cbf47ad
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 15 deletions.
diff --git a/...ain/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json b/...ain/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json
@@ -2,7 +2,7 @@
   "sourceDefinitionId": "435bb9a5-7887-4809-aa58-28c27df0d7ad",
   "name": "MySQL",
   "dockerRepository": "airbyte/source-mysql",
-  "dockerImageTag": "0.3.2",
+  "dockerImageTag": "0.3.3",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql",
   "icon": "mysql.svg"
 }
diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml
@@ -71,7 +71,7 @@
 - sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   name: MySQL
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 0.3.2
+  dockerImageTag: 0.3.3
   documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
   icon: mysql.svg
 - sourceDefinitionId: 2470e835-feaf-4db6-96f3-70fd645acc77

diff --git a/airbyte-integrations/connectors/source-mysql/Dockerfile b/airbyte-integrations/connectors/source-mysql/Dockerfile
@@ -8,6 +8,6 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.3.2
+LABEL io.airbyte.version=0.3.3
 
 LABEL io.airbyte.name=airbyte/source-mysql
diff --git a/...urce-mysql/src/main/java/io/airbyte/integrations/source/mysql/DebeziumRecordIterator.java b/...urce-mysql/src/main/java/io/airbyte/integrations/source/mysql/DebeziumRecordIterator.java
@@ -53,17 +53,14 @@ public class DebeziumRecordIterator extends AbstractIterator<ChangeEvent<String,
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumRecordIterator.class);
 
-  /**
-   * This is not private and final because we need to override in tests otherwise each test would
-   * continue to run for 5 minutes
-   */
-  static TimeUnit sleepTimeUnit = TimeUnit.MINUTES;
-  private static final int SLEEP_TIME_AMOUNT = 5;
+  private static final WaitTime FIRST_RECORD_WAIT_TIME_MINUTES = new WaitTime(5, TimeUnit.MINUTES);
+  private static final WaitTime SUBSEQUENT_RECORD_WAIT_TIME_SECONDS = new WaitTime(5, TimeUnit.SECONDS);
 
   private final LinkedBlockingQueue<ChangeEvent<String, String>> queue;
   private final Optional<TargetFilePosition> targetFilePosition;
   private final Supplier<Boolean> publisherStatusSupplier;
   private final VoidCallable requestClose;
+  private boolean receivedFirstRecord;
 
   public DebeziumRecordIterator(LinkedBlockingQueue<ChangeEvent<String, String>> queue,
                                 Optional<TargetFilePosition> targetFilePosition,
@@ -73,6 +70,7 @@ public DebeziumRecordIterator(LinkedBlockingQueue<ChangeEvent<String, String>> q
     this.targetFilePosition = targetFilePosition;
     this.publisherStatusSupplier = publisherStatusSupplier;
     this.requestClose = requestClose;
+    this.receivedFirstRecord = false;
   }
 
   @Override
@@ -83,7 +81,8 @@ protected ChangeEvent<String, String> computeNext() {
     while (!MoreBooleans.isTruthy(publisherStatusSupplier.get()) || !queue.isEmpty()) {
       final ChangeEvent<String, String> next;
       try {
-        next = queue.poll(SLEEP_TIME_AMOUNT, sleepTimeUnit);
+        WaitTime waitTime = receivedFirstRecord ? SUBSEQUENT_RECORD_WAIT_TIME_SECONDS : FIRST_RECORD_WAIT_TIME_MINUTES;
+        next = queue.poll(waitTime.period, waitTime.timeUnit);
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -100,7 +99,7 @@ protected ChangeEvent<String, String> computeNext() {
       if (shouldSignalClose(next)) {
         requestClose();
       }
-
+      receivedFirstRecord = true;
       return next;
     }
     return endOfData();
@@ -146,4 +145,16 @@ enum SnapshotMetadata {
     LAST
   }
 
+  private static class WaitTime {
+
+    public final int period;
+    public final TimeUnit timeUnit;
+
+    public WaitTime(int period, TimeUnit timeUnit) {
+      this.period = period;
+      this.timeUnit = timeUnit;
+    }
+
+  }
+
 }
diff --git a/...t-integration/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceAcceptanceTest.java b/...t-integration/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceAcceptanceTest.java
@@ -42,7 +42,6 @@
 import io.airbyte.protocol.models.SyncMode;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import org.jooq.SQLDialect;
 import org.testcontainers.containers.MySQLContainer;
 
@@ -109,7 +108,6 @@ protected List<String> getRegexTests() {
 
   @Override
   protected void setup(TestDestinationEnv testEnv) {
-    DebeziumRecordIterator.sleepTimeUnit = TimeUnit.SECONDS;
     container = new MySQLContainer<>("mysql:8.0");
     container.start();
 

diff --git a/...s/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceTest.java b/...s/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceTest.java
@@ -68,7 +68,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.jooq.SQLDialect;
@@ -132,7 +131,6 @@ public void setup() {
   }
 
   private void init() {
-    DebeziumRecordIterator.sleepTimeUnit = TimeUnit.SECONDS;
     container = new MySQLContainer<>("mysql:8.0");
     container.start();
     source = new MySqlSource();