🐛 Source MySQL CDC: sync failed when has Zero-date value in mandatory…

… column (#10251)

* replace null to epoch if column is mandatory

* format

* add tests

* incr mysql encrypt version

* bump version

* bump version

* fix changelog

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.5.2",
+  "dockerImageTag": "0.5.4",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql",
   "icon": "mysql.svg"
 }

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

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

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

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

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/internals/MySQLConverter.java

@@ -4,8 +4,10 @@
 
 package io.airbyte.integrations.debezium.internals;
 
+import io.airbyte.db.DataTypeUtils;
 import io.debezium.spi.converter.CustomConverter;
 import io.debezium.spi.converter.RelationalColumn;
+import java.time.LocalDate;
 import java.util.Arrays;
 import java.util.Properties;
 import org.apache.kafka.connect.data.SchemaBuilder;
@@ -55,9 +57,18 @@ private void registerText(final RelationalColumn field, final ConverterRegistrat
     });
   }
 
+  /**
+   * The debezium driver replaces Zero-value by Null even when this column is mandatory. According to
+   * the doc, it should be done by driver, but it fails.
+   */
+  private Object convertDefaultValueNullDate(final RelationalColumn field) {
+    var defaultValue = DebeziumConverterUtils.convertDefaultValue(field);
+    return (defaultValue == null && !field.isOptional() ? DataTypeUtils.toISO8601String(LocalDate.EPOCH) : defaultValue);
+  }
+
   private void registerDate(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
     registration.register(SchemaBuilder.string(),
-        x -> x == null ? DebeziumConverterUtils.convertDefaultValue(field) : DebeziumConverterUtils.convertDate(x));
+        x -> x == null ? convertDefaultValueNullDate(field) : DebeziumConverterUtils.convertDate(x));
   }
 
 }

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

@@ -16,5 +16,5 @@ ENV APPLICATION source-mysql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.8
+LABEL io.airbyte.version=0.1.10
 LABEL io.airbyte.name=airbyte/source-mysql-strict-encrypt

airbyte-integrations/connectors/source-mysql/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-mysql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.5.2
+LABEL io.airbyte.version=0.5.4
 LABEL io.airbyte.name=airbyte/source-mysql

airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceDatatypeTest.java

@@ -240,6 +240,16 @@ protected void initTests() {
             .addExpectedValues(null, "2021-01-01T00:00:00Z")
             .build());
 
+    // Check Zero-date value for mandatory field
+    addDataTypeTestData(
+        TestDataHolder.builder()
+             .sourceType("date")
+             .fullSourceDataType("date not null")
+             .airbyteType(JsonSchemaPrimitive.STRING)
+             .addInsertValues("'0000-00-00'")
+             .addExpectedValues("1970-01-01T00:00:00Z")
+             .build());
+
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("datetime")
@@ -248,6 +258,16 @@ protected void initTests() {
             .addExpectedValues(null, "2005-10-10T23:22:21Z")
             .build());
 
+    // Check Zero-date value for mandatory field
+    addDataTypeTestData(
+        TestDataHolder.builder()
+            .sourceType("datetime")
+            .fullSourceDataType("datetime not null")
+            .airbyteType(JsonSchemaPrimitive.STRING)
+            .addInsertValues("'0000-00-00 00:00:00'")
+            .addExpectedValues("1970-01-01T00:00:00Z")
+            .build());
+
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("timestamp")
@@ -256,6 +276,16 @@ protected void initTests() {
             .addNullExpectedValue()
             .build());
 
+    // Check Zero-date value for mandatory field
+    addDataTypeTestData(
+       TestDataHolder.builder()
+            .sourceType("timestamp")
+            .fullSourceDataType("timestamp not null")
+            .airbyteType(JsonSchemaPrimitive.STRING)
+            .addInsertValues("'0000-00-00 00:00:00.000000'")
+            .addExpectedValues("1970-01-01T00:00:00Z")
+            .build());
+
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("time")