MSSQL and debezium improvements

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/ChangeEventWithMetadata.kt

@@ -6,29 +6,22 @@ package io.airbyte.cdk.integrations.debezium.internals
 import com.fasterxml.jackson.databind.JsonNode
 import io.airbyte.commons.json.Jsons
 import io.debezium.engine.ChangeEvent
+import io.github.oshai.kotlinlogging.KotlinLogging
 
-class ChangeEventWithMetadata(private val event: ChangeEvent<String?, String?>) {
-    private val eventKeyAsJson: JsonNode = Jsons.deserialize(event.key())
-    private val eventValueAsJson: JsonNode = Jsons.deserialize(event.value())
-    private val snapshotMetadata: SnapshotMetadata? =
-        SnapshotMetadata.Companion.fromString(eventValueAsJson["source"]["snapshot"].asText())
-
-    fun event(): ChangeEvent<String?, String?> {
-        return event
-    }
+private val LOGGER = KotlinLogging.logger {}
 
-    fun eventKeyAsJson(): JsonNode {
-        return eventKeyAsJson
-    }
+class ChangeEventWithMetadata(private val event: ChangeEvent<String?, String?>) {
+    val eventKeyAsJson: JsonNode?
+        get() = event.key()?.let { Jsons.deserialize(it) }.also { it ?: LOGGER.warn { "Event key is null $event" } }
+    val eventValueAsJson: JsonNode?
+        get() = event.value()?.let { Jsons.deserialize(it) }.also { it ?: LOGGER.warn { "Event value is null $event" } }
 
-    fun eventValueAsJson(): JsonNode {
-        return eventValueAsJson
-    }
+    val snapshotMetadata: SnapshotMetadata?
+        get() {
+            val metadataKey = eventValueAsJson?.get("source")?.get("snapshot")?.asText()
+            return metadataKey?.let { return SnapshotMetadata.fromString(metadataKey) }
+        }
 
     val isSnapshotEvent: Boolean
-        get() = SnapshotMetadata.Companion.isSnapshotEventMetadata(snapshotMetadata)
-
-    fun snapshotMetadata(): SnapshotMetadata? {
-        return snapshotMetadata
-    }
+        get() = SnapshotMetadata.isSnapshotEventMetadata(snapshotMetadata)
 }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/DebeziumRecordIterator.kt

@@ -181,7 +181,7 @@ class DebeziumRecordIterator<T>(
             hasSnapshotFinished = !changeEventWithMetadata.isSnapshotEvent
 
             if (isEventLogged) {
-                val source: JsonNode? = changeEventWithMetadata.eventValueAsJson()["source"]
+                val source: JsonNode? = changeEventWithMetadata.eventValueAsJson?.get("source")
                 LOGGER.info {
                     "CDC events queue poll(): " +
                         "returned a change event with \"source\": $source."
@@ -340,8 +340,10 @@ class DebeziumRecordIterator<T>(
          * snapshots) t: truncate, m: message
          */
         fun isEventTypeHandled(event: ChangeEventWithMetadata): Boolean {
-            event.eventValueAsJson()["op"]?.asText()?.let {
-                return it in listOf("c", "u", "d", "r", "t")
+            LOGGER.info { "*** isEventTypeHandled for ${event.eventValueAsJson}" }
+            LOGGER.info { "*** op ${event.eventValueAsJson?.get("op")?.asText()}" }
+            event.eventValueAsJson?.get("op")?.asText()?.let {
+                return it in listOf("c", "u", "d", "t")
             }
                 ?: return false
         }

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/debezium/internals/RelationalDbDebeziumEventConverter.kt

@@ -14,11 +14,11 @@ class RelationalDbDebeziumEventConverter(
     private val emittedAt: Instant
 ) : DebeziumEventConverter {
     override fun toAirbyteMessage(event: ChangeEventWithMetadata): AirbyteMessage {
-        val debeziumEvent = event.eventValueAsJson()
-        val before: JsonNode? = debeziumEvent.get(DebeziumEventConverter.Companion.BEFORE_EVENT)
-        val after: JsonNode? = debeziumEvent.get(DebeziumEventConverter.Companion.AFTER_EVENT)
+        val debeziumEvent = event.eventValueAsJson
+        val before: JsonNode? = debeziumEvent?.get(DebeziumEventConverter.Companion.BEFORE_EVENT)
+        val after: JsonNode? = debeziumEvent?.get(DebeziumEventConverter.Companion.AFTER_EVENT)
         val source: JsonNode =
-            checkNotNull(debeziumEvent.get(DebeziumEventConverter.Companion.SOURCE_EVENT)) {
+            checkNotNull(debeziumEvent?.get(DebeziumEventConverter.Companion.SOURCE_EVENT)) {
                 "ChangeEvent contains no source record $debeziumEvent"
             }
 

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

@@ -5,7 +5,7 @@ plugins {
 airbyteJavaConnector {
     cdkVersionRequired = '0.44.4'
     features = ['db-sources']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 java {
@@ -25,7 +25,7 @@ application {
 
 dependencies {
     implementation 'com.microsoft.sqlserver:mssql-jdbc:12.6.1.jre11'
-    implementation 'io.debezium:debezium-embedded:2.6.1.Final'
+    implementation 'io.debezium:debezium-embedded:2.7.1.Final'
     implementation 'io.debezium:debezium-connector-sqlserver:2.6.2.Final'
     implementation 'org.codehaus.plexus:plexus-utils:3.4.2'
 

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

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 4.1.8
+  dockerImageTag: 4.1.9
   dockerRepository: airbyte/source-mssql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
   githubIssueLabel: source-mssql

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcHelper.java

@@ -83,7 +83,7 @@ public static Properties getDebeziumProperties(final JdbcDatabase database, fina
     } else {
       // If not in snapshot mode, initial will make sure that a snapshot is taken if the transaction log
       // is rotated out. This will also end up read streaming changes from the transaction_log.
-      props.setProperty("snapshot.mode", "initial");
+      props.setProperty("snapshot.mode", "recovery");
     }
 
     props.setProperty("snapshot.isolation.mode", "read_committed");

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcTargetPosition.java

@@ -35,11 +35,11 @@ public MssqlCdcTargetPosition(final Lsn targetLsn) {
   public boolean reachedTargetPosition(final ChangeEventWithMetadata changeEventWithMetadata) {
     if (changeEventWithMetadata.isSnapshotEvent()) {
       return false;
-    } else if (SnapshotMetadata.LAST == changeEventWithMetadata.snapshotMetadata()) {
+    } else if (SnapshotMetadata.LAST == changeEventWithMetadata.getSnapshotMetadata()) {
       LOGGER.info("Signalling close because Snapshot is complete");
       return true;
     } else {
-      final Lsn recordLsn = extractLsn(changeEventWithMetadata.eventValueAsJson());
+      final Lsn recordLsn = extractLsn(changeEventWithMetadata.getEventValueAsJson());
       final boolean isEventLSNAfter = targetLsn.compareTo(recordLsn) <= 0;
       if (isEventLSNAfter) {
         LOGGER.info("Signalling close because record's LSN : " + recordLsn + " is after target LSN : " + targetLsn);
@@ -122,7 +122,7 @@ public boolean isEventAheadOffset(Map<String, String> offset, ChangeEventWithMet
     if (offset == null || offset.size() != 1) {
       return false;
     }
-    final Lsn eventLsn = extractLsn(event.eventValueAsJson());
+    final Lsn eventLsn = extractLsn(event.getEventValueAsJson());
     final Lsn offsetLsn = offsetToLsn(offset);
     return eventLsn.compareTo(offsetLsn) > 0;
   }