-
Notifications
You must be signed in to change notification settings - Fork 4.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
🎉 Source MSSQL: implementation for CDC (#4689)
* first few classes for mssql cdc * wip * mssql cdc working against unit tests * increment version * add cdc acceptance test * tweaks * add file * working on comprehensive tests * change isolation from snapshot to read_committed_snapshot * finalised type tests * Revert "change isolation from snapshot to read_committed_snapshot" This reverts commit 20c6768. * small docstring fix * remove unused imports * stress test fixes * minor formatting improvements * mssql cdc docs * finish off cdc docs * format fix * update connector version * add to changelog * fix for sql server agent offline failing cdc enable on tables * final structure * few more updates * undo unwanted changes * add abstract test + more refinement * remove CDC metadata to debezium * use new cdc abstraction for mysql * undo wanted change * use cdc abstraction for postgres * add files * pull in latest changes * ready * rename class + add missing property * use renamed class + move constants to MySqlSource * use renamed class + move constants to PostgresSource * move debezium to bases + upgrade debezium version + review comments * downgrade version + minor fixes * bring in latest changes from cdc abstraction * reset to minutes * bring in the latest changes * format * fix build * address review comments * bring in latest changes * bring in latest changes * use common abstraction for CDC via debezium for sql server * remove debezium from build * finalise PR * should return Optional * pull in latest changes * pull in latest changes * address review comments * use common abstraction for CDC via debezium for mysql (#4604) * use new cdc abstraction for mysql * undo wanted change * pull in latest changes * use renamed class + move constants to MySqlSource * bring in latest changes from cdc abstraction * format * bring in latest changes * pull in latest changes * use common abstraction for CDC via debezium for postgres (#4607) * use cdc abstraction for postgres * add files * ready * use renamed class + move constants to PostgresSource * bring in the latest changes * bring in latest changes * pull in latest changes * lower version for tests to run on CI * format * Update docs/integrations/sources/mssql.md Co-authored-by: Sherif A. Nada <snadalive@gmail.com> * addressing review comments * fix for testGetTargetPosition * format changes Co-authored-by: George Claireaux <george@claireaux.co.uk> Co-authored-by: Sherif A. Nada <snadalive@gmail.com>
- Loading branch information
3 people
authored
Jul 14, 2021
1 parent
b18cedd
commit cc7b376
Showing
19 changed files
with
1,933 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
46 changes: 46 additions & 0 deletions
46
...src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcConnectorMetadataInjector.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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(); | ||
| } | ||
|
|
||
| } |
52 changes: 52 additions & 0 deletions
52
...s/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcProperties.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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; | ||
| } | ||
|
|
||
| } |
56 changes: 56 additions & 0 deletions
56
...ce-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcSavedInfoFetcher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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); | ||
| } | ||
|
|
||
| } |
68 changes: 68 additions & 0 deletions
68
...source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcStateHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.