Patch for mssql source cdc update

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

@@ -16,5 +16,5 @@ ENV APPLICATION source-mssql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.22
+LABEL io.airbyte.version=0.4.0
 LABEL io.airbyte.name=airbyte/source-mssql-strict-encrypt

airbyte-integrations/connectors/source-mssql-strict-encrypt/src/test/resources/expected_spec.json

@@ -90,13 +90,61 @@
           }
         ]
       },
-      "replication_method": {
-        "type": "string",
+      "replication": {
+        "type": "object",
         "title": "Replication Method",
         "description": "The replication method used for extracting data from the database. STANDARD replication requires no setup on the DB side but will not be able to represent deletions incrementally. CDC uses {TBC} to detect inserts, updates, and deletes. This needs to be configured on the source database itself.",
         "default": "STANDARD",
-        "enum": ["STANDARD", "CDC"],
-        "order": 8
+        "additionalProperties": true,
+        "order": 8,
+        "oneOf": [
+          {
+            "title": "Standard",
+            "additionalProperties": false,
+            "description": "Standard replication requires no setup on the DB side but will not be able to represent deletions incrementally.",
+            "required": ["replication_type"],
+            "properties": {
+              "replication_type": {
+                "type": "string",
+                "const": "Standard",
+                "enum": ["Standard"],
+                "default": "Standard",
+                "order": 0
+              }
+            }
+          },
+          {
+            "title": "Logical Replication (CDC)",
+            "additionalProperties": false,
+            "description": "CDC uses {TBC} to detect inserts, updates, and deletes. This needs to be configured on the source database itself.",
+            "required": ["replication_type"],
+            "properties": {
+              "replication_type": {
+                "type": "string",
+                "const": "CDC",
+                "enum": ["CDC"],
+                "default": "CDC",
+                "order":0
+              },
+              "data_to_sync": {
+                "title": "Data to Sync",
+                "type": "string",
+                "default": "Existing and New",
+                "enum": ["Existing and New", "New Changes Only"],
+                "description": "What data should be synced under the CDC. \"Existing and New\" will read existing data as a snapshot, and sync new changes through CDC. \"New Changes Only\" will skip the initial snapshot, and only sync new changes through CDC.",
+                "order": 1
+              },
+              "snapshot_isolation": {
+                "title": "Initial Snapshot Isolation Level",
+                "type": "string",
+                "default": "Snapshot",
+                "enum": ["Snapshot", "Read Committed"],
+                "description": "Existing data in the database are synced through an initial snapshot. This parameter controls the isolation level that will be used during the initial snapshotting. If you choose the \"Snapshot\" level, you must enable the <a href=\"https://docs.microsoft.com/en-us/dotnet/framework/data/adonet/sql/snapshot-isolation-in-sql-server\">snapshot isolation mode</a> on the database.",
+                "order": 2
+              }
+            }
+          }
+        ]
       }
     }
   }

airbyte-integrations/connectors/source-mssql/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-mssql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.23
+LABEL io.airbyte.version=0.4.0
 LABEL io.airbyte.name=airbyte/source-mssql

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

@@ -0,0 +1,158 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.source.mssql;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.debezium.annotation.VisibleForTesting;
+import java.util.Properties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MssqlCdcHelper {
+
+  // legacy replication method config before version 0.4.0
+  // it is an enum with possible values: STANDARD and CDC
+  private static final String LEGACY_REPLICATION_FIELD = "replication_method";
+  // new replication method config since version 0.4.0
+  // it is an oneOf object
+  private static final String REPLICATION_FIELD = "replication";
+  private static final String REPLICATION_TYPE_FIELD = "replication_type";
+  private static final String CDC_SNAPSHOT_ISOLATION_FIELD = "snapshot_isolation";
+  private static final String CDC_DATA_TO_SYNC_FIELD = "data_to_sync";
+
+  public enum ReplicationMethod {
+    STANDARD,
+    CDC
+  }
+
+  /**
+   * The default "SNAPSHOT" mode can prevent other (non-Airbyte) transactions from updating table rows
+   * while we snapshot. References:
+   * https://docs.microsoft.com/en-us/sql/t-sql/statements/set-transaction-isolation-level-transact-sql?view=sql-server-ver15
+   * https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-isolation-mode
+   */
+  public enum SnapshotIsolation {
+
+    SNAPSHOT("Snapshot", "snapshot"),
+    READ_COMMITTED("Read Committed", "read_committed");
+
+    private final String snapshotIsolationLevel;
+    private final String debeziumIsolationMode;
+
+    SnapshotIsolation(final String snapshotIsolationLevel, final String debeziumIsolationMode) {
+      this.snapshotIsolationLevel = snapshotIsolationLevel;
+      this.debeziumIsolationMode = debeziumIsolationMode;
+    }
+
+    public String getDebeziumIsolationMode() {
+      return debeziumIsolationMode;
+    }
+
+    public static SnapshotIsolation from(final String jsonValue) {
+      for (final SnapshotIsolation value : values()) {
+        if (value.snapshotIsolationLevel.equalsIgnoreCase(jsonValue)) {
+          return value;
+        }
+      }
+      throw new IllegalArgumentException("Unexpected snapshot isolation level: " + jsonValue);
+    }
+
+  }
+
+  // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-mode
+  public enum DataToSync {
+
+    EXISTING_AND_NEW("Existing and New", "initial"),
+    NEW_CHANGES_ONLY("New Changes Only", "schema_only");
+
+    private final String dataToSyncConfig;
+    private final String debeziumSnapshotMode;
+
+    DataToSync(final String value, final String debeziumSnapshotMode) {
+      this.dataToSyncConfig = value;
+      this.debeziumSnapshotMode = debeziumSnapshotMode;
+    }
+
+    public String getDebeziumSnapshotMode() {
+      return debeziumSnapshotMode;
+    }
+
+    public static DataToSync from(final String value) {
+      for (final DataToSync s : values()) {
+        if (s.dataToSyncConfig.equalsIgnoreCase(value)) {
+          return s;
+        }
+      }
+      throw new IllegalArgumentException("Unexpected data to sync setting: " + value);
+    }
+
+  }
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(MssqlCdcHelper.class);
+
+  @VisibleForTesting
+  static boolean isCdc(final JsonNode config) {
+    // legacy replication method config before version 0.4.0
+    if (config.hasNonNull(LEGACY_REPLICATION_FIELD)) {
+      return ReplicationMethod.valueOf(config.get(LEGACY_REPLICATION_FIELD).asText()) == ReplicationMethod.CDC;
+    }
+    // new replication method config since version 0.4.0
+    if (config.hasNonNull(REPLICATION_FIELD)) {
+      final JsonNode replicationConfig = config.get(REPLICATION_FIELD);
+      return ReplicationMethod.valueOf(replicationConfig.get(REPLICATION_TYPE_FIELD).asText()) == ReplicationMethod.CDC;
+    }
+    return false;
+  }
+
+  @VisibleForTesting
+  static SnapshotIsolation getSnapshotIsolationConfig(final JsonNode config) {
+    // legacy replication method config before version 0.4.0
+    if (config.hasNonNull(LEGACY_REPLICATION_FIELD)) {
+      return SnapshotIsolation.SNAPSHOT;
+    }
+    // new replication method config since version 0.4.0
+    if (config.hasNonNull(REPLICATION_FIELD)) {
+      final JsonNode replicationConfig = config.get(REPLICATION_FIELD);
+      final JsonNode snapshotIsolation = replicationConfig.get(CDC_SNAPSHOT_ISOLATION_FIELD);
+      return SnapshotIsolation.from(snapshotIsolation.asText());
+    }
+    return SnapshotIsolation.SNAPSHOT;
+  }
+
+  @VisibleForTesting
+  static DataToSync getDataToSyncConfig(final JsonNode config) {
+    // legacy replication method config before version 0.4.0
+    if (config.hasNonNull(LEGACY_REPLICATION_FIELD)) {
+      return DataToSync.EXISTING_AND_NEW;
+    }
+    // new replication method config since version 0.4.0
+    if (config.hasNonNull(REPLICATION_FIELD)) {
+      final JsonNode replicationConfig = config.get(REPLICATION_FIELD);
+      final JsonNode dataToSync = replicationConfig.get(CDC_DATA_TO_SYNC_FIELD);
+      return DataToSync.from(dataToSync.asText());
+    }
+    return DataToSync.EXISTING_AND_NEW;
+  }
+
+  @VisibleForTesting
+  static Properties getDebeziumProperties(final JsonNode config) {
+    final Properties props = new Properties();
+    props.setProperty("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
+
+    // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-include-schema-changes
+    props.setProperty("include.schema.changes", "false");
+    // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-provide-transaction-metadata
+    props.setProperty("provide.transaction.metadata", "false");
+
+    props.setProperty("converters", "mssql_converter");
+    props.setProperty("mssql_converter.type", "io.airbyte.integrations.debezium.internals.MSSQLConverter");
+
+    props.setProperty("snapshot.mode", getDataToSyncConfig(config).getDebeziumSnapshotMode());
+    props.setProperty("snapshot.isolation.mode", getSnapshotIsolationConfig(config).getDebeziumIsolationMode());
+
+    return props;
+  }
+
+}