mysql-cdc:increase duration for records from debezium from 5 seconds …

…to 5 minutes (#3757) * mysql-cdc:increase duration for records from debezium from 5 seconds to 5 minutes * upgrade docker version
airbytehq · Jun 1, 2021 · 9179a65 · 9179a65
1 parent 1fed4b0
commit 9179a65
Show file tree

Hide file tree

Showing 6 changed files with 13 additions and 5 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.1",
+  "dockerImageTag": "0.3.2",
   "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
@@ -66,7 +66,7 @@
 - sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   name: MySQL
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 0.3.1
+  dockerImageTag: 0.3.2
   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.1
+LABEL io.airbyte.version=0.3.2
 
 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,7 +53,11 @@ public class DebeziumRecordIterator extends AbstractIterator<ChangeEvent<String,
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumRecordIterator.class);
 
-  private static final TimeUnit SLEEP_TIME_UNIT = TimeUnit.SECONDS;
+  /**
+   * 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 final LinkedBlockingQueue<ChangeEvent<String, String>> queue;
@@ -79,7 +83,7 @@ protected ChangeEvent<String, String> computeNext() {
     while (!MoreBooleans.isTruthy(publisherStatusSupplier.get()) || !queue.isEmpty()) {
       final ChangeEvent<String, String> next;
       try {
-        next = queue.poll(SLEEP_TIME_AMOUNT, SLEEP_TIME_UNIT);
+        next = queue.poll(SLEEP_TIME_AMOUNT, sleepTimeUnit);
       } catch (InterruptedException e) {
         throw new RuntimeException(e);
       }

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,6 +42,7 @@
 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;
 
@@ -108,6 +109,7 @@ 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,6 +68,7 @@
 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;
@@ -131,6 +132,7 @@ public void setup() {
   }
 
   private void init() {
+    DebeziumRecordIterator.sleepTimeUnit = TimeUnit.SECONDS;
     container = new MySQLContainer<>("mysql:8.0");
     container.start();
     source = new MySqlSource();