Prevent conversion of null fields

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.kt

@@ -43,6 +43,15 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
             val columnName = queryContext.metaData.getColumnName(i)
             val columnTypeName = queryContext.metaData.getColumnTypeName(i)
             try {
+                // attempt to access the column. this allows us to know if it is null before we do
+                // type-specific
+                // parsing. if it is null, we can move on. while awkward, this seems to be the
+                // agreed upon way of checking for null values with jdbc.
+                queryContext.getObject(i)
+                if (queryContext.wasNull()) {
+                    continue
+                }
+
                 // convert to java types that will convert into reasonable json.
                 copyToJsonField(queryContext, i, jsonNode)
             } catch (e: java.lang.Exception) {

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.29.13
+version=0.29.14

airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/source/AbstractSourceDatabaseTypeTest.kt

@@ -28,8 +28,8 @@ import org.slf4j.LoggerFactory
  * type system.
  */
 abstract class AbstractSourceDatabaseTypeTest : AbstractSourceConnectorTest() {
-    protected val testDataHolders: MutableList<TestDataHolder> = ArrayList()
-    protected var database: Database? = null
+    @JvmField protected val testDataHolders: MutableList<TestDataHolder> = ArrayList()
+    @JvmField protected var database: Database? = null
 
     protected val idColumnName: String
         /**

airbyte-integrations/connectors/source-mssql/build.gradle

@@ -3,9 +3,9 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.29.1'
+    cdkVersionRequired = '0.29.14'
     features = ['db-sources']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 java {

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.8
+  dockerImageTag: 4.0.9
   dockerRepository: airbyte/source-mssql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
   githubIssueLabel: source-mssql
@@ -19,12 +19,8 @@ data:
   name: Microsoft SQL Server (MSSQL)
   registries:
     cloud:
-      # A regression in https://github.com/airbytehq/airbyte-internal-issues/issues/7055
-      dockerImageTag: 4.0.6
       enabled: true
     oss:
-      # A regression in https://github.com/airbytehq/airbyte-internal-issues/issues/7055
-      dockerImageTag: 4.0.6
       enabled: true
   releaseStage: generally_available
   supportLevel: certified

airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/CdcMssqlSourceAcceptanceTest.java

@@ -4,9 +4,13 @@
 
 package io.airbyte.integrations.source.mssql;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static io.airbyte.protocol.models.v0.SyncMode.FULL_REFRESH;
+import static io.airbyte.protocol.models.v0.SyncMode.INCREMENTAL;
+import static org.junit.jupiter.api.Assertions.*;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import io.airbyte.cdk.integrations.base.ssh.SshHelpers;
@@ -28,7 +32,9 @@
 import io.airbyte.protocol.models.v0.DestinationSyncMode;
 import io.airbyte.protocol.models.v0.SyncMode;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
+import org.junit.Assert;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.api.TestInstance.Lifecycle;
@@ -43,6 +49,7 @@ public class CdcMssqlSourceAcceptanceTest extends SourceAcceptanceTest {
   private static final String STREAM_NAME = "id_and_name";
   private static final String STREAM_NAME2 = "starships";
   private static final String CDC_ROLE_NAME = "cdc_selector";
+  private static final String STREAM_NAME3 = "stream3";
 
   private MsSQLTestDatabase testdb;
 
@@ -111,12 +118,15 @@ protected void setupEnvironment(final TestDestinationEnv environment) {
         // create tables
         .with("CREATE TABLE %s.%s(id INTEGER PRIMARY KEY, name VARCHAR(200));", SCHEMA_NAME, STREAM_NAME)
         .with("CREATE TABLE %s.%s(id INTEGER PRIMARY KEY, name VARCHAR(200));", SCHEMA_NAME, STREAM_NAME2)
+        .with("CREATE TABLE %s.%s (id INTEGER PRIMARY KEY, name VARCHAR(200), userid INTEGER DEFAULT NULL);", SCHEMA_NAME, STREAM_NAME3)
         // populate tables
         .with("INSERT INTO %s.%s (id, name) VALUES (1,'picard'),  (2, 'crusher'), (3, 'vash');", SCHEMA_NAME, STREAM_NAME)
         .with("INSERT INTO %s.%s (id, name) VALUES (1,'enterprise-d'),  (2, 'defiant'), (3, 'yamato');", SCHEMA_NAME, STREAM_NAME2)
+        .with("INSERT INTO %s.%s (id, name) VALUES (4,'voyager');", SCHEMA_NAME, STREAM_NAME3)
         // enable cdc on tables for designated role
         .withCdcForTable(SCHEMA_NAME, STREAM_NAME, CDC_ROLE_NAME)
         .withCdcForTable(SCHEMA_NAME, STREAM_NAME2, CDC_ROLE_NAME)
+        .withCdcForTable(SCHEMA_NAME, STREAM_NAME3, CDC_ROLE_NAME)
         // revoke user permissions
         .with("REVOKE ALL FROM %s CASCADE;", testdb.getUserName())
         .with("EXEC sp_msforeachtable \"REVOKE ALL ON '?' TO %s;\"", testdb.getUserName())
@@ -177,4 +187,79 @@ private List<AirbyteStateMessage> filterStateMessages(final List<AirbyteMessage>
         .collect(Collectors.toList());
   }
 
+  @Test
+  protected void testNullValueConversion() throws Exception {
+    final List<ConfiguredAirbyteStream> configuredAirbyteStreams =
+        Lists.newArrayList(new ConfiguredAirbyteStream()
+            .withSyncMode(INCREMENTAL)
+            .withDestinationSyncMode(DestinationSyncMode.APPEND)
+            .withStream(CatalogHelpers.createAirbyteStream(STREAM_NAME3,
+                SCHEMA_NAME,
+                Field.of("id", JsonSchemaType.NUMBER),
+                Field.of("name", JsonSchemaType.STRING),
+                Field.of("userid", JsonSchemaType.NUMBER))
+                .withSourceDefinedCursor(true)
+                .withSourceDefinedPrimaryKey(List.of(List.of("id")))
+                .withSupportedSyncModes(Lists.newArrayList(FULL_REFRESH, INCREMENTAL))));
+
+    final ConfiguredAirbyteCatalog configuredCatalogWithOneStream =
+        new ConfiguredAirbyteCatalog().withStreams(List.of(configuredAirbyteStreams.get(0)));
+
+    final List<AirbyteMessage> airbyteMessages = runRead(configuredCatalogWithOneStream, getState());
+    final List<AirbyteRecordMessage> recordMessages = filterRecords(airbyteMessages);
+    final List<AirbyteStateMessage> stateMessages = airbyteMessages
+        .stream()
+        .filter(m -> m.getType() == AirbyteMessage.Type.STATE)
+        .map(AirbyteMessage::getState)
+        .collect(Collectors.toList());
+    Assert.assertEquals(recordMessages.size(), 1);
+    assertFalse(stateMessages.isEmpty(), "Reason");
+    // TODO validate exact records
+    ObjectMapper mapper = new ObjectMapper();
+
+    assertTrue(cdcFieldsOmitted(recordMessages.get(0).getData()).equals(mapper.readTree("""
+                                                                                        {"id":4, "name":"voyager"}""")));
+
+    // when we run incremental sync again there should be no new records. Run a sync with the latest
+    // state message and assert no records were emitted.
+    JsonNode latestState = null;
+    for (final AirbyteStateMessage stateMessage : stateMessages) {
+      if (stateMessage.getType().equals(AirbyteStateMessage.AirbyteStateType.STREAM)) {
+        latestState = Jsons.jsonNode(stateMessages);
+        break;
+      } else if (stateMessage.getType().equals(AirbyteStateMessage.AirbyteStateType.GLOBAL)) {
+        latestState = Jsons.jsonNode(List.of(Iterables.getLast(stateMessages)));
+        break;
+      } else {
+        throw new RuntimeException("Unknown state type " + stateMessage.getType());
+      }
+    }
+
+    testdb.getDatabase().query(c -> {
+      return c.query("""
+                     INSERT INTO %s.%s (id, name) VALUES (5,'deep space nine');
+                     """.formatted(SCHEMA_NAME, STREAM_NAME3));
+    }).execute();
+
+    assert Objects.nonNull(latestState);
+    final List<AirbyteRecordMessage> secondSyncRecords = filterRecords(runRead(configuredCatalogWithOneStream, latestState));
+    assertFalse(
+        secondSyncRecords.isEmpty(),
+        "Expected the second incremental sync to produce records.");
+    assertTrue(cdcFieldsOmitted(secondSyncRecords.get(0).getData()).equals(mapper.readTree("""
+                                                                                           {"id":5, "name":"deep space nine", "userid":null}""")));
+
+  }
+
+  private JsonNode cdcFieldsOmitted(final JsonNode node) {
+    ObjectMapper mapper = new ObjectMapper();
+    ObjectNode object = mapper.createObjectNode();
+    node.fieldNames().forEachRemaining(name -> {
+      if (!name.toLowerCase().startsWith("_ab_cdc_")) {
+        object.put(name, node.get(name));
+      }
+    });
+    return object;
+  }
+
 }

airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/MssqlSourceAcceptanceTest.java

@@ -4,9 +4,13 @@
 
 package io.airbyte.integrations.source.mssql;
 
+import static io.airbyte.protocol.models.v0.SyncMode.INCREMENTAL;
 import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import io.airbyte.cdk.integrations.base.ssh.SshHelpers;
@@ -29,6 +33,7 @@
 import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
 
@@ -43,7 +48,7 @@ public class MssqlSourceAcceptanceTest extends SourceAcceptanceTest {
   @Override
   protected void setupEnvironment(final TestDestinationEnv environment) throws SQLException {
     testdb = MsSQLTestDatabase.in(BaseImage.MSSQL_2022)
-        .with("CREATE TABLE id_and_name (id INTEGER, name VARCHAR(200), born DATETIMEOFFSET(7));")
+        .with("CREATE TABLE %s.%s (id INTEGER, name VARCHAR(200), born DATETIMEOFFSET(7));", SCHEMA_NAME, STREAM_NAME)
         .with("CREATE TABLE %s.%s(id INTEGER PRIMARY KEY, name VARCHAR(200));", SCHEMA_NAME, STREAM_NAME2)
         .with("INSERT INTO id_and_name (id, name, born) VALUES " +
             "(1, 'picard', '2124-03-04T01:01:01Z'), " +
@@ -76,12 +81,26 @@ protected JsonNode getConfig() {
 
   @Override
   protected ConfiguredAirbyteCatalog getConfiguredCatalog() {
-    return CatalogHelpers.createConfiguredAirbyteCatalog(
-        STREAM_NAME,
-        SCHEMA_NAME,
-        Field.of("id", JsonSchemaType.NUMBER),
-        Field.of("name", JsonSchemaType.STRING),
-        Field.of("born", JsonSchemaType.STRING));
+    return new ConfiguredAirbyteCatalog().withStreams(Lists.newArrayList(
+        new ConfiguredAirbyteStream()
+            .withSyncMode(INCREMENTAL)
+            .withCursorField(Lists.newArrayList("id"))
+            .withDestinationSyncMode(DestinationSyncMode.APPEND)
+            .withStream(CatalogHelpers.createAirbyteStream(
+                String.format("%s", STREAM_NAME), SCHEMA_NAME,
+                Field.of("id", JsonSchemaType.NUMBER),
+                Field.of("name", JsonSchemaType.STRING))
+                .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, INCREMENTAL))),
+        new ConfiguredAirbyteStream()
+            .withSyncMode(INCREMENTAL)
+            .withCursorField(Lists.newArrayList("id"))
+            .withDestinationSyncMode(DestinationSyncMode.APPEND)
+            .withStream(CatalogHelpers.createAirbyteStream(
+                String.format("%s", STREAM_NAME2),
+                SCHEMA_NAME,
+                Field.of("id", JsonSchemaType.NUMBER),
+                Field.of("name", JsonSchemaType.STRING))
+                .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, INCREMENTAL)))));
   }
 
   @Override
@@ -97,14 +116,14 @@ protected void testAddNewStreamToExistingSync() throws Exception {
             Field.of("id", JsonSchemaType.NUMBER),
             Field.of("name", JsonSchemaType.STRING))
             .withDestinationSyncMode(DestinationSyncMode.APPEND)
-            .withSyncMode(SyncMode.INCREMENTAL)
+            .withSyncMode(INCREMENTAL)
             .withCursorField(List.of("id")),
             CatalogHelpers.createConfiguredAirbyteStream(STREAM_NAME2,
                 SCHEMA_NAME,
                 Field.of("id", JsonSchemaType.NUMBER),
                 Field.of("name", JsonSchemaType.STRING))
                 .withDestinationSyncMode(DestinationSyncMode.APPEND)
-                .withSyncMode(SyncMode.INCREMENTAL)
+                .withSyncMode(INCREMENTAL)
                 .withCursorField(List.of("id")));
     final ConfiguredAirbyteCatalog configuredCatalogWithOneStream =
         new ConfiguredAirbyteCatalog().withStreams(List.of(configuredAirbyteStreams.get(0)));
@@ -139,6 +158,79 @@ protected void testAddNewStreamToExistingSync() throws Exception {
     assertEquals(SCHEMA_NAME, stateMessages2.get(1).getStream().getStreamDescriptor().getNamespace());
   }
 
+  @Test
+  protected void testNullValueConversion() throws Exception {
+    final String STREAM_NAME3 = "stream3";
+    testdb.getDatabase().query(c -> {
+      return c.query("""
+                     CREATE TABLE %s.%s (id INTEGER PRIMARY KEY, name VARCHAR(200), userid INTEGER DEFAULT NULL);
+                     """.formatted(SCHEMA_NAME, STREAM_NAME3));
+    }).execute();
+
+    testdb.getDatabase().query(c -> {
+      return c.query("""
+                     INSERT INTO %s.%s (id, name) VALUES (4,'voyager');
+                     """.formatted(SCHEMA_NAME, STREAM_NAME3));
+    }).execute();
+
+    final List<ConfiguredAirbyteStream> configuredAirbyteStreams =
+        Lists.newArrayList(CatalogHelpers.createConfiguredAirbyteStream(STREAM_NAME3,
+            SCHEMA_NAME,
+            Field.of("id", JsonSchemaType.NUMBER),
+            Field.of("name", JsonSchemaType.STRING),
+            Field.of("userid", JsonSchemaType.NUMBER))
+            .withDestinationSyncMode(DestinationSyncMode.APPEND)
+            .withSyncMode(INCREMENTAL)
+            .withCursorField(List.of("id")));
+    final ConfiguredAirbyteCatalog configuredCatalogWithOneStream =
+        new ConfiguredAirbyteCatalog().withStreams(List.of(configuredAirbyteStreams.get(0)));
+
+    final List<AirbyteMessage> airbyteMessages = runRead(configuredCatalogWithOneStream, getState());
+    final List<AirbyteRecordMessage> recordMessages = filterRecords(airbyteMessages);
+    final List<AirbyteStateMessage> stateMessages = airbyteMessages
+        .stream()
+        .filter(m -> m.getType() == AirbyteMessage.Type.STATE)
+        .map(AirbyteMessage::getState)
+        .collect(Collectors.toList());
+    assertEquals(recordMessages.size(), 1);
+    assertFalse(stateMessages.isEmpty(), "Reason");
+    // TODO validate exact records
+    ObjectMapper mapper = new ObjectMapper();
+
+    assertTrue(recordMessages.get(0).getData().equals(mapper.readTree("""
+                                                                      {"id":4, "name":"voyager"}""")));
+
+    // when we run incremental sync again there should be no new records. Run a sync with the latest
+    // state message and assert no records were emitted.
+    JsonNode latestState = null;
+    for (final AirbyteStateMessage stateMessage : stateMessages) {
+      if (stateMessage.getType().equals(AirbyteStateMessage.AirbyteStateType.STREAM)) {
+        latestState = Jsons.jsonNode(stateMessages);
+        break;
+      } else if (stateMessage.getType().equals(AirbyteStateMessage.AirbyteStateType.GLOBAL)) {
+        latestState = Jsons.jsonNode(List.of(Iterables.getLast(stateMessages)));
+        break;
+      } else {
+        throw new RuntimeException("Unknown state type " + stateMessage.getType());
+      }
+    }
+
+    testdb.getDatabase().query(c -> {
+      return c.query("""
+                     INSERT INTO %s.%s (id, name) VALUES (5,'deep space nine');
+                     """.formatted(SCHEMA_NAME, STREAM_NAME3));
+    }).execute();
+
+    assert Objects.nonNull(latestState);
+    final List<AirbyteRecordMessage> secondSyncRecords = filterRecords(runRead(configuredCatalogWithOneStream, latestState));
+    assertFalse(
+        secondSyncRecords.isEmpty(),
+        "Expected the second incremental sync to produce records.");
+    assertTrue(secondSyncRecords.get(0).getData().equals(mapper.readTree("""
+                                                                         {"id":5, "name":"deep space nine"}""")));
+
+  }
+
   private List<AirbyteStateMessage> filterStateMessages(final List<AirbyteMessage> messages) {
     return messages.stream().filter(r -> r.getType() == AirbyteMessage.Type.STATE).map(AirbyteMessage::getState)
         .collect(Collectors.toList());

airbyte-integrations/connectors/source-mysql/build.gradle

@@ -6,9 +6,9 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.29.6'
+    cdkVersionRequired = '0.29.14'
     features = ['db-sources']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 java {

airbyte-integrations/connectors/source-mysql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
-  dockerImageTag: 3.3.16
+  dockerImageTag: 3.3.17
   dockerRepository: airbyte/source-mysql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   githubIssueLabel: source-mysql
@@ -19,12 +19,8 @@ data:
   name: MySQL
   registries:
     cloud:
-      # A regression in https://github.com/airbytehq/airbyte-internal-issues/issues/7055
-      dockerImageTag: 3.3.13
       enabled: true
     oss:
-      # A regression in https://github.com/airbytehq/airbyte-internal-issues/issues/7055
-      dockerImageTag: 3.3.13
       enabled: true
   releaseStage: generally_available
   releases: