-
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.
🎉 Postgres source: sync data from beginning if lsn is no longer valid…
… in cdc (#15077) * work in progress * cleanup * add test * introduce tests for state parsing util class * enable test via feature flag * review comments * Bump versions * auto-bump connector version [ci skip] Co-authored-by: Liren Tu <tuliren.git@outlook.com> Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
- Loading branch information
3 people
authored
Aug 3, 2022
1 parent
d57a6a6
commit f4b4863
Showing
13 changed files
with
499 additions
and
111 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
102 changes: 102 additions & 0 deletions
102
...2/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumPropertiesManager.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,102 @@ | ||
| /* | ||
| * Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.debezium.internals; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.db.jdbc.JdbcUtils; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteStream; | ||
| import io.airbyte.protocol.models.SyncMode; | ||
| import java.util.Optional; | ||
| import java.util.Properties; | ||
| import java.util.stream.Collectors; | ||
| import org.codehaus.plexus.util.StringUtils; | ||
|
|
||
| public class DebeziumPropertiesManager { | ||
|
|
||
| private final JsonNode config; | ||
| private final AirbyteFileOffsetBackingStore offsetManager; | ||
| private final Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager; | ||
|
|
||
| private final Properties properties; | ||
| private final ConfiguredAirbyteCatalog catalog; | ||
|
|
||
| public DebeziumPropertiesManager(final Properties properties, | ||
| final JsonNode config, | ||
| final ConfiguredAirbyteCatalog catalog, | ||
| final AirbyteFileOffsetBackingStore offsetManager, | ||
| final Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager) { | ||
| this.properties = properties; | ||
| this.config = config; | ||
| this.catalog = catalog; | ||
| this.offsetManager = offsetManager; | ||
| this.schemaHistoryManager = schemaHistoryManager; | ||
| } | ||
|
|
||
| protected Properties getDebeziumProperties() { | ||
| final Properties props = new Properties(); | ||
| props.putAll(properties); | ||
|
|
||
| // debezium engine configuration | ||
| props.setProperty("name", "engine"); | ||
| props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore"); | ||
| props.setProperty("offset.storage.file.filename", offsetManager.getOffsetFilePath().toString()); | ||
| props.setProperty("offset.flush.interval.ms", "1000"); // todo: make this longer | ||
| // default values from debezium CommonConnectorConfig | ||
| props.setProperty("max.batch.size", "2048"); | ||
| props.setProperty("max.queue.size", "8192"); | ||
|
|
||
| if (schemaHistoryManager.isPresent()) { | ||
| // https://debezium.io/documentation/reference/1.9/operations/debezium-server.html#debezium-source-database-history-class | ||
| // https://debezium.io/documentation/reference/development/engine.html#_in_the_code | ||
| // As mentioned in the documents above, debezium connector for MySQL needs to track the schema | ||
| // changes. If we don't do this, we can't fetch records for the table. | ||
| props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory"); | ||
| props.setProperty("database.history.file.filename", schemaHistoryManager.get().getPath().toString()); | ||
| } | ||
|
|
||
| // https://debezium.io/documentation/reference/configuration/avro.html | ||
| props.setProperty("key.converter.schemas.enable", "false"); | ||
| props.setProperty("value.converter.schemas.enable", "false"); | ||
|
|
||
| // debezium names | ||
| props.setProperty("name", config.get(JdbcUtils.DATABASE_KEY).asText()); | ||
| props.setProperty("database.server.name", config.get(JdbcUtils.DATABASE_KEY).asText()); | ||
|
|
||
| // db connection configuration | ||
| props.setProperty("database.hostname", config.get(JdbcUtils.HOST_KEY).asText()); | ||
| props.setProperty("database.port", config.get(JdbcUtils.PORT_KEY).asText()); | ||
| props.setProperty("database.user", config.get(JdbcUtils.USERNAME_KEY).asText()); | ||
| props.setProperty("database.dbname", config.get(JdbcUtils.DATABASE_KEY).asText()); | ||
|
|
||
| if (config.has(JdbcUtils.PASSWORD_KEY)) { | ||
| props.setProperty("database.password", config.get(JdbcUtils.PASSWORD_KEY).asText()); | ||
| } | ||
|
|
||
| // By default "decimal.handing.mode=precise" which's caused returning this value as a binary. | ||
| // The "double" type may cause a loss of precision, so set Debezium's config to store it as a String | ||
| // explicitly in its Kafka messages for more details see: | ||
| // https://debezium.io/documentation/reference/1.9/connectors/postgresql.html#postgresql-decimal-types | ||
| // https://debezium.io/documentation/faq/#how_to_retrieve_decimal_field_from_binary_representation | ||
| props.setProperty("decimal.handling.mode", "string"); | ||
|
|
||
| // table selection | ||
| final String tableWhitelist = getTableWhitelist(catalog); | ||
| props.setProperty("table.include.list", tableWhitelist); | ||
|
|
||
| return props; | ||
| } | ||
|
|
||
| public static String getTableWhitelist(final ConfiguredAirbyteCatalog catalog) { | ||
| return catalog.getStreams().stream() | ||
| .filter(s -> s.getSyncMode() == SyncMode.INCREMENTAL) | ||
| .map(ConfiguredAirbyteStream::getStream) | ||
| .map(stream -> stream.getNamespace() + "." + stream.getName()) | ||
| // debezium needs commas escaped to split properly | ||
| .map(x -> StringUtils.escape(x, new char[] {','}, "\\,")) | ||
| .collect(Collectors.joining(",")); | ||
| } | ||
|
|
||
| } |
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
32 changes: 32 additions & 0 deletions
32
...v1-9-2/src/main/java/io/airbyte/integrations/debezium/internals/PostgresCustomLoader.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,32 @@ | ||
| /* | ||
| * Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.debezium.internals; | ||
|
|
||
| import io.airbyte.commons.json.Jsons; | ||
| import io.debezium.connector.postgresql.PostgresConnectorConfig; | ||
| import io.debezium.connector.postgresql.PostgresOffsetContext; | ||
| import io.debezium.connector.postgresql.PostgresOffsetContext.Loader; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
|
|
||
| public class PostgresCustomLoader extends Loader { | ||
|
|
||
| private Map<String, ?> offset; | ||
|
|
||
| public PostgresCustomLoader(PostgresConnectorConfig connectorConfig) { | ||
| super(connectorConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public PostgresOffsetContext load(Map<String, ?> offset) { | ||
| this.offset = Jsons.clone(offset); | ||
| return super.load(offset); | ||
| } | ||
|
|
||
| public Map<String, ?> getRawOffset() { | ||
| return Collections.unmodifiableMap(offset); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.