🐛 Source-MySql: do not check cdc required param binlog_row_image for standard replication

airbyte-config/init/src/main/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.4.12",
+  "dockerImageTag": "0.4.13",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql",
   "icon": "mysql.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -414,7 +414,7 @@
 - name: MySQL
   sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 0.4.11
+  dockerImageTag: 0.4.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
   icon: mysql.svg
   sourceType: database

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -3963,7 +3963,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mysql:0.4.11"
+- dockerImage: "airbyte/source-mysql:0.4.13"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/mysql"
     connectionSpecification:

airbyte-integrations/connectors/source-mysql-strict-encrypt/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.4
+LABEL io.airbyte.version=0.1.5
 LABEL io.airbyte.name=airbyte/source-mysql-strict-encrypt

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.4.12
+LABEL io.airbyte.version=0.4.13
 
 LABEL io.airbyte.name=airbyte/source-mysql

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java

@@ -97,58 +97,10 @@ private static AirbyteStream addCdcMetadataColumns(final AirbyteStream stream) {
   public List<CheckedConsumer<JdbcDatabase, Exception>> getCheckOperations(final JsonNode config) throws Exception {
     final List<CheckedConsumer<JdbcDatabase, Exception>> checkOperations = new ArrayList<>(super.getCheckOperations(config));
     if (isCdc(config)) {
-      checkOperations.add(database -> {
-        final List<String> log = database.resultSetQuery(connection -> {
-          final String sql = "show variables where Variable_name = 'log_bin'";
-
-          return connection.createStatement().executeQuery(sql);
-        }, resultSet -> resultSet.getString("Value")).collect(toList());
-
-        if (log.size() != 1) {
-          throw new RuntimeException("Could not query the variable log_bin");
-        }
-
-        final String logBin = log.get(0);
-        if (!logBin.equalsIgnoreCase("ON")) {
-          throw new RuntimeException("The variable log_bin should be set to ON, but it is : " + logBin);
-        }
-      });
-
-      checkOperations.add(database -> {
-        final List<String> format = database.resultSetQuery(connection -> {
-          final String sql = "show variables where Variable_name = 'binlog_format'";
-
-          return connection.createStatement().executeQuery(sql);
-        }, resultSet -> resultSet.getString("Value")).collect(toList());
-
-        if (format.size() != 1) {
-          throw new RuntimeException("Could not query the variable binlog_format");
-        }
-
-        final String binlogFormat = format.get(0);
-        if (!binlogFormat.equalsIgnoreCase("ROW")) {
-          throw new RuntimeException("The variable binlog_format should be set to ROW, but it is : " + binlogFormat);
-        }
-      });
+      checkOperations.addAll(List.of(getCheckOperation("log_bin", "ON"),
+          getCheckOperation("binlog_format", "ROW"),
+          getCheckOperation("binlog_row_image", "FULL")));
     }
-
-    checkOperations.add(database -> {
-      final List<String> image = database.resultSetQuery(connection -> {
-        final String sql = "show variables where Variable_name = 'binlog_row_image'";
-
-        return connection.createStatement().executeQuery(sql);
-      }, resultSet -> resultSet.getString("Value")).collect(toList());
-
-      if (image.size() != 1) {
-        throw new RuntimeException("Could not query the variable binlog_row_image");
-      }
-
-      final String binlogRowImage = image.get(0);
-      if (!binlogRowImage.equalsIgnoreCase("FULL")) {
-        throw new RuntimeException("The variable binlog_row_image should be set to FULL, but it is : " + binlogRowImage);
-      }
-    });
-
     return checkOperations;
   }
 
@@ -253,4 +205,23 @@ public enum ReplicationMethod {
     CDC
   }
 
+  private CheckedConsumer<JdbcDatabase, Exception> getCheckOperation(String name, String value) {
+    return database -> {
+      final List<String> result = database.resultSetQuery(connection -> {
+        final String sql = String.format("show variables where Variable_name = '%s'", name);
+
+        return connection.createStatement().executeQuery(sql);
+      }, resultSet -> resultSet.getString("Value")).collect(toList());
+
+      if (result.size() != 1) {
+        throw new RuntimeException(String.format("Could not query the variable %s", name));
+      }
+
+      final String resultValue = result.get(0);
+      if (!resultValue.equalsIgnoreCase(value)) {
+        throw new RuntimeException(String.format("The variable %s should be set to %s, but it is : %s", name, value, resultValue));
+      }
+    };
+  }
+
 }

docs/integrations/sources/mysql.md

@@ -180,7 +180,8 @@ If you do not see a type in this list, assume that it is coerced into a string.
 
 | Version | Date | Pull Request | Subject |
 |:--------| :--- | :--- | :--- |
-| 0.3.12  | 2021-12-01 | [8371](https://github.com/airbytehq/airbyte/pull/8371) | Fixed incorrect handling "\n" in ssh key |
+| 0.4.13  | 2021-12-03 | [8335](https://github.com/airbytehq/airbyte/pull/8335) | Source-MySql: do not check cdc required param binlog_row_image for standard replication |
+| 0.4.12  | 2021-12-01 | [8371](https://github.com/airbytehq/airbyte/pull/8371) | Fixed incorrect handling "\n" in ssh key |
 | 0.4.11  | 2021-11-19 | [8047](https://github.com/airbytehq/airbyte/pull/8047) | Source MySQL: transform binary data base64 format |
 | 0.4.10  | 2021-11-15 | [7820](https://github.com/airbytehq/airbyte/pull/7820) | Added basic performance test |
 | 0.4.9   | 2021-11-02 | [7559](https://github.com/airbytehq/airbyte/pull/7559) | Correctly process large unsigned short integer values which may fall outside java's `Short` data type capability |