🐛 source-mssql: connector forces snapshot isolation level

[sh4sh]

Connector Name

source-mssql

Connector Version

1.1.0

What step the error happened?

Configuring a new connector

Revelant information

PR with proposed fix: #28545

There appears to be a bug in source-mssql's support for CDC + Read Committed. Currently both snapshot isolation levels check the db for enabled full Snapshot Isolation, which it should not do when the Read Committed isolation level is selected.

Support for non-snapshot CDC is added here: #13168
In the connector spec we see a rework of replication_method to support the Read Committed isolation level. replication_method field is renamed to replication for hygiene.
We can see from comments that the new implementation since 0.4.0 should use the replication parameter instead of the replication_method. More recently however, the connector was modified in #16215 to use replication_method which has caused some problems.

In the original change to support Read Committed snapshot isolation level for MSSQL + CDC, we see the new REPLICATION_FIELD value pulled into CdcMssqlHelper.java:

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

Lines 15 to 20 in 4284ce6

	// 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";

isCdc checks whether CDC is enabled. It supports both LEGACY_REPLICATION_FIELD and REPLICATION_FIELD:

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

Lines 95 to 107 in 4284ce6

	@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;
	}

getSnapshotIsolationConfig checks the snapshot isolation level:

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

Lines 109 to 122 in 4284ce6

	@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;
	}

Since then, the spec parameter was modified from replication back to replication_method. This means that the getSnapshotIsolationConfig function still needs REPLICATION_FIELD to be hasNonNull. Unfortunately however it's always null, because REPLICATION_FIELD doesn't exist anymore, we only have LEGACY_REPLICATION_FIELD.

So now, if we look at master branch, we observe that the connector config is not passed into some functions.
i.e.,

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

Lines 120 to 140 in 7e4797d

	@VisibleForTesting
	static SnapshotIsolation getSnapshotIsolationConfig(final JsonNode config) {
	// 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) {
	// 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;
	}

tl;dr quick/non-breaking-change fix is to update functions with the correct config parameter

Since this work was originally done to make naming consistent throughout CDC connectors (a reasonable goal) we should probably look at refactoring.

Relevant log output

No response

Contribute

• Yes, I want to contribute