configure debezium to correctly convert column with default value whe… (#36011)

rodireich · web-flow · commit fd98267c4b62 · 2024-03-13T11:34:43.000-07:00
diff --git a/airbyte-cdk/java/airbyte-cdk/README.md b/airbyte-cdk/java/airbyte-cdk/README.md
@@ -166,6 +166,7 @@ MavenLocal debugging steps:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.23.20 | 2024-03-12 | [\#36011](https://github.com/airbytehq/airbyte/pull/36011) | Debezium configuration for conversion of null value on a column with default value.                                                                            |
 | 0.23.19 | 2024-03-11 | [\#35904](https://github.com/airbytehq/airbyte/pull/35904) | Add retries to the debezium engine.                                                                                                                            |
 | 0.23.18 | 2024-03-07 | [\#35899](https://github.com/airbytehq/airbyte/pull/35899) | Null check when retrieving destination state                                                                                                                   |
 | 0.23.16 | 2024-03-06 | [\#35842](https://github.com/airbytehq/airbyte/pull/35842) | Improve logging in debezium processing.                                                                                                                        |
diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
@@ -1 +1 @@
-version=0.23.19
+version=0.23.20
diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumPropertiesManager.java
@@ -76,7 +76,11 @@ public Properties getDebeziumProperties(
     // WARNING : Never change the value of this otherwise all the connectors would start syncing from
     // scratch.
     props.setProperty(TOPIC_PREFIX_KEY, sanitizeTopicPrefix(getName(config)));
-
+    // https://issues.redhat.com/browse/DBZ-7635
+    // https://cwiki.apache.org/confluence/display/KAFKA/KIP-581%3A+Value+of+optional+null+field+which+has+default+value
+    // A null value in a column with default value won't be generated correctly in CDC unless we set the
+    // following
+    props.setProperty("value.converter.replace.null.with.default", "false");
     // includes
     props.putAll(getIncludeConfiguration(catalog, config));
 
diff --git a/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/SnapshotMetadata.java b/airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/internals/SnapshotMetadata.java
@@ -16,7 +16,8 @@ public enum SnapshotMetadata {
   LAST_IN_DATA_COLLECTION,
   TRUE,
   LAST,
-  FALSE;
+  FALSE,
+  NULL;
 
   private static final Set<SnapshotMetadata> ENTRIES_OF_SNAPSHOT_EVENTS =
       ImmutableSet.of(TRUE, FIRST, FIRST_IN_DATA_COLLECTION, LAST_IN_DATA_COLLECTION);
@@ -35,6 +36,8 @@ public enum SnapshotMetadata {
     STRING_TO_ENUM.put("LAST_IN_DATA_COLLECTION", LAST_IN_DATA_COLLECTION);
     STRING_TO_ENUM.put("first_in_data_collection", FIRST_IN_DATA_COLLECTION);
     STRING_TO_ENUM.put("FIRST_IN_DATA_COLLECTION", FIRST_IN_DATA_COLLECTION);
+    STRING_TO_ENUM.put("NULL", NULL);
+    STRING_TO_ENUM.put("null", NULL);
   }
 
   public static SnapshotMetadata fromString(final String value) {
diff --git a/airbyte-integrations/connectors/source-mssql/build.gradle b/airbyte-integrations/connectors/source-mssql/build.gradle
@@ -3,7 +3,7 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.23.15'
+    cdkVersionRequired = '0.23.20'
     features = ['db-sources']
     useLocalCdk = false
 }
diff --git a/airbyte-integrations/connectors/source-mssql/metadata.yaml b/airbyte-integrations/connectors/source-mssql/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 4.0.0
+  dockerImageTag: 4.0.1
   dockerRepository: airbyte/source-mssql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
   githubIssueLabel: source-mssql
diff --git a/airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/AbstractMssqlSourceDatatypeTest.java b/airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/AbstractMssqlSourceDatatypeTest.java
@@ -344,6 +344,14 @@ protected void initTests() {
 
     }
 
+    addDataTypeTestData(
+        TestDataHolder.builder()
+            .sourceType("int")
+            .airbyteType(JsonSchemaType.INTEGER)
+            .addInsertValues("null", "1234", "7878")
+            .addExpectedValues(null, "1234", "7878")
+            .createTablePatternSql("CREATE TABLE %1$s(%2$s INTEGER NULL DEFAULT ((7878)), %3$s %4$s)")
+            .build());
   }
 
 }
diff --git a/docs/integrations/sources/mssql.md b/docs/integrations/sources/mssql.md
@@ -327,6 +327,7 @@ WHERE actor_definition_id ='b5ea17b1-f170-46dc-bc31-cc744ca984c1' AND (configura
 
 | Version | Date       | Pull Request                                                                                                      | Subject                                                                                                                                         |
 |:--------|:-----------|:------------------------------------------------------------------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------|
+| 4.0.1   | 2024-03-12 | [36011](https://github.com/airbytehq/airbyte/pull/36011)                                                               | Read correctly null values of columns with default value in CDC.                                                                                |
 | 4.0.0   | 2024-03-06 | [35873](https://github.com/airbytehq/airbyte/pull/35873)                                                          | Terabyte-sized tables support, reliability improvements, bug fixes.                                                                             |
 | 3.7.7   | 2024-03-06 | [35816](https://github.com/airbytehq/airbyte/pull/35816)                                                          | Fix query that was failing on a case sensitive server.                                                                                          |
 | 3.7.6   | 2024-03-04 | [35721](https://github.com/airbytehq/airbyte/pull/35721)                                                          | Fix tests                                                                                                                                       |

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@ plugins {`
`3`	`3`	`}`
`4`	`4`
`5`	`5`	`airbyteJavaConnector {`
`6`		`- cdkVersionRequired = '0.23.15'`
	`6`	`+ cdkVersionRequired = '0.23.20'`
`7`	`7`	`features = ['db-sources']`
`8`	`8`	`useLocalCdk = false`
`9`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -344,6 +344,14 @@ protected void initTests() {`
`344`	`344`
`345`	`345`	`}`
`346`	`346`
	`347`	`+ addDataTypeTestData(`
	`348`	`+ TestDataHolder.builder()`
	`349`	`+ .sourceType("int")`
	`350`	`+ .airbyteType(JsonSchemaType.INTEGER)`
	`351`	`+ .addInsertValues("null", "1234", "7878")`
	`352`	`+ .addExpectedValues(null, "1234", "7878")`
	`353`	`+ .createTablePatternSql("CREATE TABLE %1$s(%2$s INTEGER NULL DEFAULT ((7878)), %3$s %4$s)")`
	`354`	`+ .build());`
`347`	`355`	`}`
`348`	`356`
`349`	`357`	`}`