Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🎉 Source MSSQL: implementation for CDC #4689

Merged
merged 88 commits into from
Jul 14, 2021
Merged
Show file tree
Hide file tree
Changes from 83 commits
Commits
Show all changes
88 commits
Select commit Hold shift + click to select a range
4eb28aa
first few classes for mssql cdc
Phlair Jun 25, 2021
0495779
wip
subodh1810 Jun 29, 2021
08d86fe
mssql cdc working against unit tests
Phlair Jun 30, 2021
bd70631
increment version
Phlair Jun 30, 2021
d3f44b3
add cdc acceptance test
Phlair Jun 30, 2021
418cd3e
tweaks
Phlair Jun 30, 2021
fbc52cd
add file
subodh1810 Jul 1, 2021
b178706
working on comprehensive tests
Phlair Jul 1, 2021
ac1b4ab
Merge branch 'master' into george/sql-server-cdc
Phlair Jul 1, 2021
20c6768
change isolation from snapshot to read_committed_snapshot
Phlair Jul 2, 2021
bb167d0
finalised type tests
Phlair Jul 2, 2021
ed9b322
Revert "change isolation from snapshot to read_committed_snapshot"
Phlair Jul 2, 2021
7a2e6b2
small docstring fix
Phlair Jul 4, 2021
5f765b2
remove unused imports
Phlair Jul 4, 2021
551bb55
stress test fixes
Phlair Jul 4, 2021
7a385b0
minor formatting improvements
Phlair Jul 4, 2021
5fbeaad
mssql cdc docs
Phlair Jul 4, 2021
a0df5a6
Merge branch 'master' into george/sql-server-cdc
Phlair Jul 4, 2021
216179a
finish off cdc docs
Phlair Jul 5, 2021
a2fc97e
format fix
Phlair Jul 5, 2021
eafad5c
update connector version
Phlair Jul 5, 2021
c842577
add to changelog
Phlair Jul 5, 2021
7845443
fix for sql server agent offline failing cdc enable on tables
Phlair Jul 5, 2021
3d34b9b
final structure
subodh1810 Jul 5, 2021
ec59346
few more updates
subodh1810 Jul 6, 2021
557fb7a
undo unwanted changes
subodh1810 Jul 6, 2021
071a445
add abstract test + more refinement
subodh1810 Jul 7, 2021
8fa0e5e
remove CDC metadata to debezium
subodh1810 Jul 7, 2021
0ab4cc3
use new cdc abstraction for mysql
subodh1810 Jul 7, 2021
1d76edc
undo wanted change
subodh1810 Jul 7, 2021
433d424
use cdc abstraction for postgres
subodh1810 Jul 7, 2021
53f88be
add files
subodh1810 Jul 7, 2021
eaf3b0b
Merge branch 'master' into cdc-abstraction
subodh1810 Jul 7, 2021
15340c9
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 7, 2021
deeee6d
pull in latest changes
subodh1810 Jul 7, 2021
97d95f1
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 7, 2021
8179081
ready
subodh1810 Jul 7, 2021
8ea4e99
Merge branch 'master' into cdc-abstraction
subodh1810 Jul 7, 2021
0b7581b
rename class + add missing property
subodh1810 Jul 7, 2021
8bded3b
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 7, 2021
4ae7a7c
use renamed class + move constants to MySqlSource
subodh1810 Jul 7, 2021
423ed0d
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 7, 2021
c33c8d8
use renamed class + move constants to PostgresSource
subodh1810 Jul 7, 2021
356c884
Merge branch 'master' into cdc-abstraction
subodh1810 Jul 8, 2021
f164eb0
move debezium to bases + upgrade debezium version + review comments
subodh1810 Jul 8, 2021
8ed621f
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 8, 2021
1547ab0
downgrade version + minor fixes
subodh1810 Jul 8, 2021
57b5d9b
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 8, 2021
557c354
bring in latest changes from cdc abstraction
subodh1810 Jul 8, 2021
eca4c7a
Merge branch 'master' into cdc-abstraction
subodh1810 Jul 8, 2021
924598e
reset to minutes
subodh1810 Jul 8, 2021
1b44320
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 8, 2021
75b6e55
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 8, 2021
0a8d554
bring in the latest changes
subodh1810 Jul 8, 2021
79f1f21
format
subodh1810 Jul 8, 2021
92f1f83
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 8, 2021
eb97469
Merge branch 'master' into george/sql-server-cdc
subodh1810 Jul 8, 2021
a6f7236
fix build
subodh1810 Jul 8, 2021
6ce6fd8
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 8, 2021
dd073b0
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 8, 2021
4826d0b
Merge branch 'master' into cdc-abstraction
subodh1810 Jul 12, 2021
c16076d
address review comments
subodh1810 Jul 12, 2021
733d369
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 12, 2021
fb10951
bring in latest changes
subodh1810 Jul 12, 2021
d879a34
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 12, 2021
1fba343
bring in latest changes
subodh1810 Jul 12, 2021
75a0340
Merge branch 'postgres-cdc-abstraction' into george/sql-server-cdc
subodh1810 Jul 12, 2021
61adb37
use common abstraction for CDC via debezium for sql server
subodh1810 Jul 12, 2021
00af585
remove debezium from build
subodh1810 Jul 12, 2021
f180e74
finalise PR
subodh1810 Jul 12, 2021
e542b89
should return Optional
subodh1810 Jul 12, 2021
831a606
Merge branch 'cdc-abstraction' into use-cdc-abstraction-for-mysql
subodh1810 Jul 12, 2021
b0f8b2e
pull in latest changes
subodh1810 Jul 12, 2021
161ad4c
Merge branch 'use-cdc-abstraction-for-mysql' into postgres-cdc-abstra…
subodh1810 Jul 12, 2021
cc7650d
pull in latest changes
subodh1810 Jul 12, 2021
cf5ea57
Merge branch 'postgres-cdc-abstraction' into new-cdc-abstraction-sql-…
subodh1810 Jul 12, 2021
ad56dca
address review comments
subodh1810 Jul 12, 2021
fbc318f
use common abstraction for CDC via debezium for mysql (#4604)
subodh1810 Jul 12, 2021
80408be
Merge branch 'cdc-abstraction' into new-cdc-abstraction-sql-server
subodh1810 Jul 12, 2021
6bb0948
lower version for tests to run on CI
subodh1810 Jul 12, 2021
36a18d4
Merge branch 'master' into new-cdc-abstraction-sql-server
subodh1810 Jul 12, 2021
323ff30
format
subodh1810 Jul 12, 2021
6898d9b
Merge branch 'master' into new-cdc-abstraction-sql-server
subodh1810 Jul 12, 2021
12e6ed0
Update docs/integrations/sources/mssql.md
Phlair Jul 13, 2021
2f87d62
addressing review comments
Phlair Jul 13, 2021
087ff64
fix for testGetTargetPosition
Phlair Jul 14, 2021
a1c04b3
Merge branch 'master' into new-cdc-abstraction-sql-server
Phlair Jul 14, 2021
fcf5cf8
format changes
subodh1810 Jul 14, 2021
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions airbyte-integrations/bases/debezium/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ dependencies {
implementation 'io.debezium:debezium-embedded:1.4.2.Final'
implementation 'io.debezium:debezium-connector-mysql:1.4.2.Final'
implementation 'io.debezium:debezium-connector-postgres:1.4.2.Final'
implementation 'io.debezium:debezium-connector-sqlserver:1.4.2.Final'

testFixturesImplementation project(':airbyte-db')
testFixturesImplementation project(':airbyte-integrations:bases:base-java')
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
Expand All @@ -78,9 +80,8 @@ public abstract class CdcSourceTest {
protected static final String COL_ID = "id";
protected static final String COL_MAKE_ID = "make_id";
protected static final String COL_MODEL = "model";
protected static final String DB_NAME = MODELS_SCHEMA;

private static final AirbyteCatalog CATALOG = new AirbyteCatalog().withStreams(List.of(
protected static final AirbyteCatalog CATALOG = new AirbyteCatalog().withStreams(List.of(
CatalogHelpers.createAirbyteStream(
MODELS_STREAM_NAME,
MODELS_SCHEMA,
Expand Down Expand Up @@ -124,6 +125,24 @@ protected void executeQuery(String query) {
}
}

public String columnClause(Map<String, String> columnsWithDataType, Optional<String> primaryKey) {
StringBuilder columnClause = new StringBuilder();
int i = 0;
for (Map.Entry<String, String> column : columnsWithDataType.entrySet()) {
columnClause.append(column.getKey());
columnClause.append(" ");
columnClause.append(column.getValue());
if (i < (columnsWithDataType.size() - 1)) {
columnClause.append(",");
columnClause.append(" ");
}
i++;
}
primaryKey.ifPresent(s -> columnClause.append(", PRIMARY KEY (").append(s).append(")"));

return columnClause.toString();
}

public void createTable(String schemaName, String tableName, String columnClause) {
executeQuery(createTableQuery(schemaName, tableName, columnClause));
}
Expand All @@ -143,7 +162,7 @@ public String createSchemaQuery(String schemaName) {
private void createAndPopulateActualTable() {
createSchema(MODELS_SCHEMA);
createTable(MODELS_SCHEMA, MODELS_STREAM_NAME,
String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200), PRIMARY KEY (%s)", COL_ID, COL_MAKE_ID, COL_MODEL, COL_ID));
columnClause(ImmutableMap.of(COL_ID, "INTEGER", COL_MAKE_ID, "INTEGER", COL_MODEL, "VARCHAR(200)"), Optional.of(COL_ID)));
for (JsonNode recordJson : MODEL_RECORDS) {
writeModelRecord(recordJson);
}
Expand All @@ -156,9 +175,8 @@ private void createAndPopulateActualTable() {
private void createAndPopulateRandomTable() {
createSchema(MODELS_SCHEMA + "_random");
createTable(MODELS_SCHEMA + "_random", MODELS_STREAM_NAME + "_random",
String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200), PRIMARY KEY (%s)", COL_ID + "_random",
COL_MAKE_ID + "_random",
COL_MODEL + "_random", COL_ID + "_random"));
columnClause(ImmutableMap.of(COL_ID + "_random", "INTEGER", COL_MAKE_ID + "_random", "INTEGER", COL_MODEL + "_random", "VARCHAR(200)"),
Optional.of(COL_ID + "_random")));
final List<JsonNode> MODEL_RECORDS_RANDOM = ImmutableList.of(
Jsons
.jsonNode(ImmutableMap
Expand Down Expand Up @@ -448,7 +466,7 @@ void testCdcAndFullRefreshInSameSync() throws Exception {
Jsons.jsonNode(ImmutableMap.of(COL_ID, 160, COL_MAKE_ID, 2, COL_MODEL, "E 350-2")));

createTable(MODELS_SCHEMA, MODELS_STREAM_NAME + "_2",
String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200), PRIMARY KEY (%s)", COL_ID, COL_MAKE_ID, COL_MODEL, COL_ID));
columnClause(ImmutableMap.of(COL_ID, "INTEGER", COL_MAKE_ID, "INTEGER", COL_MODEL, "VARCHAR(200)"), Optional.of(COL_ID)));

for (JsonNode recordJson : MODEL_RECORDS_2) {
writeRecords(recordJson, MODELS_SCHEMA, MODELS_STREAM_NAME + "_2", COL_ID,
Expand Down Expand Up @@ -571,7 +589,8 @@ void testDiscover() throws Exception {
protected AirbyteCatalog expectedCatalogForDiscover() {
final AirbyteCatalog expectedCatalog = Jsons.clone(CATALOG);

createTable(MODELS_SCHEMA, MODELS_STREAM_NAME + "_2", String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200)", COL_ID, COL_MAKE_ID, COL_MODEL));
createTable(MODELS_SCHEMA, MODELS_STREAM_NAME + "_2",
columnClause(ImmutableMap.of(COL_ID, "INTEGER", COL_MAKE_ID, "INTEGER", COL_MODEL, "VARCHAR(200)"), Optional.empty()));

List<AirbyteStream> streams = expectedCatalog.getStreams();
// stream with PK
Expand All @@ -588,7 +607,19 @@ protected AirbyteCatalog expectedCatalogForDiscover() {
streamWithoutPK.setSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
addCdcMetadataColumns(streamWithoutPK);

AirbyteStream randomStream = CatalogHelpers.createAirbyteStream(
MODELS_STREAM_NAME + "_random",
MODELS_SCHEMA + "_random",
Field.of(COL_ID + "_random", JsonSchemaPrimitive.NUMBER),
Field.of(COL_MAKE_ID + "_random", JsonSchemaPrimitive.NUMBER),
Field.of(COL_MODEL + "_random", JsonSchemaPrimitive.STRING))
.withSourceDefinedCursor(true)
.withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
.withSourceDefinedPrimaryKey(List.of(List.of(COL_ID + "_random")));
addCdcMetadataColumns(randomStream);

streams.add(streamWithoutPK);
streams.add(randomStream);
expectedCatalog.withStreams(streams);
return expectedCatalog;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -117,8 +117,9 @@ public TestDataHolderBuilder airbyteType(JsonSchemaPrimitive airbyteType) {

/**
* Set custom the create table script pattern. Use it if you source uses untypical table creation
* sql. Default patter described {@link #DEFAULT_CREATE_TABLE_SQL} Note! The patter should contains
* two String place holders for the table name and data type.
* sql. Default patter described {@link #DEFAULT_CREATE_TABLE_SQL} Note! The patter should contain
* four String place holders for the: - namespace.table name (as one placeholder together) - id
* column name - test column name - test column data type
*
* @param createTablePatternSql creation table sql pattern
* @return builder
Expand Down
4 changes: 4 additions & 0 deletions airbyte-integrations/connectors/source-mssql/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -13,18 +13,22 @@ dependencies {

implementation project(':airbyte-db')
implementation project(':airbyte-integrations:bases:base-java')
implementation project(':airbyte-integrations:bases:debezium')
implementation project(':airbyte-protocol:models')
implementation project(':airbyte-integrations:connectors:source-jdbc')
implementation project(':airbyte-integrations:connectors:source-relational-db')

implementation 'io.debezium:debezium-connector-sqlserver:1.4.2.Final'
implementation 'com.microsoft.sqlserver:mssql-jdbc:8.4.1.jre14'

testImplementation testFixtures(project(':airbyte-integrations:bases:debezium'))
testImplementation testFixtures(project(':airbyte-integrations:connectors:source-jdbc'))

testImplementation 'org.apache.commons:commons-lang3:3.11'
testImplementation "org.testcontainers:mssqlserver:1.15.1"

integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-source-test')
integrationTestJavaImplementation project(':airbyte-integrations:connectors:source-mssql')

implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
integrationTestJavaImplementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
/*
* MIT License
*
* Copyright (c) 2020 Airbyte
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/

package io.airbyte.integrations.source.mssql;

import static io.airbyte.integrations.source.mssql.MssqlSource.CDC_LSN;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.airbyte.integrations.debezium.CdcMetadataInjector;

public class MssqlCdcConnectorMetadataInjector implements CdcMetadataInjector {

@Override
public void addMetaData(ObjectNode event, JsonNode source) {
String commitLsn = source.get("commit_lsn").asText();
event.put(CDC_LSN, commitLsn);
}

@Override
public String namespace(JsonNode source) {
return source.get("schema").asText();
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
/*
* MIT License
*
* Copyright (c) 2020 Airbyte
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/

package io.airbyte.integrations.source.mssql;

import java.util.Properties;

public class MssqlCdcProperties {

static Properties getDebeziumProperties() {
final Properties props = new Properties();
props.setProperty("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");

// snapshot config
// https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-snapshot-mode
props.setProperty("snapshot.mode", "initial");
// 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
// we set this to avoid preventing other (non-Airbyte) transactions from updating table rows while
// we snapshot
props.setProperty("snapshot.isolation.mode", "snapshot");

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

return props;
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
/*
* MIT License
*
* Copyright (c) 2020 Airbyte
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/

package io.airbyte.integrations.source.mssql;

import static io.airbyte.integrations.source.mssql.MssqlSource.MSSQL_CDC_OFFSET;
import static io.airbyte.integrations.source.mssql.MssqlSource.MSSQL_DB_HISTORY;

import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.integrations.debezium.CdcSavedInfoFetcher;
import io.airbyte.integrations.source.relationaldb.models.CdcState;
import java.util.Optional;

public class MssqlCdcSavedInfoFetcher implements CdcSavedInfoFetcher {

private final JsonNode savedOffset;
private final JsonNode savedSchemaHistory;

protected MssqlCdcSavedInfoFetcher(CdcState savedState) {
final boolean savedStatePresent = savedState != null && savedState.getState() != null;
this.savedOffset = savedStatePresent ? savedState.getState().get(MSSQL_CDC_OFFSET) : null;
this.savedSchemaHistory = savedStatePresent ? savedState.getState().get(MSSQL_DB_HISTORY) : null;
}

@Override
public JsonNode getSavedOffset() {
return savedOffset;
}

@Override
public Optional<JsonNode> getSavedSchemaHistory() {
return Optional.ofNullable(savedSchemaHistory);
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
/*
* MIT License
*
* Copyright (c) 2020 Airbyte
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/

package io.airbyte.integrations.source.mssql;

import static io.airbyte.integrations.source.mssql.MssqlSource.MSSQL_CDC_OFFSET;
import static io.airbyte.integrations.source.mssql.MssqlSource.MSSQL_DB_HISTORY;

import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.commons.json.Jsons;
import io.airbyte.integrations.debezium.CdcStateHandler;
import io.airbyte.integrations.source.relationaldb.StateManager;
import io.airbyte.integrations.source.relationaldb.models.CdcState;
import io.airbyte.protocol.models.AirbyteMessage;
import io.airbyte.protocol.models.AirbyteMessage.Type;
import io.airbyte.protocol.models.AirbyteStateMessage;
import java.util.HashMap;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MssqlCdcStateHandler implements CdcStateHandler {

private static final Logger LOGGER = LoggerFactory.getLogger(MssqlCdcStateHandler.class);
private final StateManager stateManager;

public MssqlCdcStateHandler(StateManager stateManager) {
this.stateManager = stateManager;
}

@Override
public AirbyteMessage saveState(Map<String, String> offset, String dbHistory) {
Map<String, Object> state = new HashMap<>();
state.put(MSSQL_CDC_OFFSET, offset);
state.put(MSSQL_DB_HISTORY, dbHistory);

final JsonNode asJson = Jsons.jsonNode(state);

LOGGER.info("debezium state: {}", asJson);

final CdcState cdcState = new CdcState().withState(asJson);
stateManager.getCdcStateManager().setCdcState(cdcState);
final AirbyteStateMessage stateMessage = stateManager.emit();
return new AirbyteMessage().withType(Type.STATE).withState(stateMessage);
}

}
Loading