forked from airbytehq/airbyte
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
convert CDK db-sources submodules to kotlin (airbytehq#36438)
convert CDK db-sources submodules to kotlin fix compiler warnings
- Loading branch information
1 parent
430ad73
commit ca4f115
Showing
190 changed files
with
16,433 additions
and
13,625 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
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
2 changes: 1 addition & 1 deletion
2
airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
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 |
|---|---|---|
| @@ -1 +1 @@ | ||
| version=0.28.10 | ||
| version=0.28.11 |
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
148 changes: 1 addition & 147 deletions
148
...db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/AirbyteDebeziumHandler.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 |
|---|---|---|
| @@ -1,149 +1,3 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.integrations.debezium; | ||
|
|
||
| import static io.airbyte.cdk.integrations.debezium.DebeziumIteratorConstants.*; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.cdk.db.jdbc.JdbcUtils; | ||
| import io.airbyte.cdk.integrations.debezium.internals.*; | ||
| import io.airbyte.cdk.integrations.source.relationaldb.state.SourceStateIterator; | ||
| import io.airbyte.cdk.integrations.source.relationaldb.state.StateEmitFrequency; | ||
| import io.airbyte.commons.util.AutoCloseableIterator; | ||
| import io.airbyte.commons.util.AutoCloseableIterators; | ||
| import io.airbyte.protocol.models.v0.AirbyteMessage; | ||
| import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog; | ||
| import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream; | ||
| import io.airbyte.protocol.models.v0.SyncMode; | ||
| import io.debezium.engine.ChangeEvent; | ||
| import io.debezium.engine.DebeziumEngine; | ||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.time.temporal.ChronoUnit; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.LinkedBlockingQueue; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * This class acts as the bridge between Airbyte DB connectors and debezium. If a DB connector wants | ||
| * to use debezium for CDC, it should use this class | ||
| */ | ||
| public class AirbyteDebeziumHandler<T> { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(AirbyteDebeziumHandler.class); | ||
| /** | ||
| * We use 10000 as capacity cause the default queue size and batch size of debezium is : | ||
| * {@link io.debezium.config.CommonConnectorConfig#DEFAULT_MAX_BATCH_SIZE}is 2048 | ||
| * {@link io.debezium.config.CommonConnectorConfig#DEFAULT_MAX_QUEUE_SIZE} is 8192 | ||
| */ | ||
| public static final int QUEUE_CAPACITY = 10_000; | ||
|
|
||
| private final JsonNode config; | ||
| private final CdcTargetPosition<T> targetPosition; | ||
| private final boolean trackSchemaHistory; | ||
| private final Duration firstRecordWaitTime, subsequentRecordWaitTime; | ||
| private final int queueSize; | ||
| private final boolean addDbNameToOffsetState; | ||
|
|
||
| public AirbyteDebeziumHandler(final JsonNode config, | ||
| final CdcTargetPosition<T> targetPosition, | ||
| final boolean trackSchemaHistory, | ||
| final Duration firstRecordWaitTime, | ||
| final Duration subsequentRecordWaitTime, | ||
| final int queueSize, | ||
| final boolean addDbNameToOffsetState) { | ||
| this.config = config; | ||
| this.targetPosition = targetPosition; | ||
| this.trackSchemaHistory = trackSchemaHistory; | ||
| this.firstRecordWaitTime = firstRecordWaitTime; | ||
| this.subsequentRecordWaitTime = subsequentRecordWaitTime; | ||
| this.queueSize = queueSize; | ||
| this.addDbNameToOffsetState = addDbNameToOffsetState; | ||
| } | ||
|
|
||
| class CapacityReportingBlockingQueue<E> extends LinkedBlockingQueue<E> { | ||
|
|
||
| private static Duration REPORT_DURATION = Duration.of(10, ChronoUnit.SECONDS); | ||
| private Instant lastReport; | ||
|
|
||
| CapacityReportingBlockingQueue(final int capacity) { | ||
| super(capacity); | ||
| } | ||
|
|
||
| private void reportQueueUtilization() { | ||
| if (lastReport == null || Duration.between(lastReport, Instant.now()).compareTo(REPORT_DURATION) > 0) { | ||
| LOGGER.info("CDC events queue size: {}. remaining {}", this.size(), this.remainingCapacity()); | ||
| synchronized (this) { | ||
| lastReport = Instant.now(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void put(final E e) throws InterruptedException { | ||
| reportQueueUtilization(); | ||
| super.put(e); | ||
| } | ||
|
|
||
| @Override | ||
| public E poll() { | ||
| reportQueueUtilization(); | ||
| return super.poll(); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| public AutoCloseableIterator<AirbyteMessage> getIncrementalIterators(final DebeziumPropertiesManager debeziumPropertiesManager, | ||
| final DebeziumEventConverter eventConverter, | ||
| final CdcSavedInfoFetcher cdcSavedInfoFetcher, | ||
| final CdcStateHandler cdcStateHandler) { | ||
| LOGGER.info("Using CDC: {}", true); | ||
| LOGGER.info("Using DBZ version: {}", DebeziumEngine.class.getPackage().getImplementationVersion()); | ||
| final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeState( | ||
| cdcSavedInfoFetcher.getSavedOffset(), | ||
| addDbNameToOffsetState ? Optional.ofNullable(config.get(JdbcUtils.DATABASE_KEY).asText()) : Optional.empty()); | ||
| final var schemaHistoryManager = trackSchemaHistory | ||
| ? Optional.of(AirbyteSchemaHistoryStorage.initializeDBHistory( | ||
| cdcSavedInfoFetcher.getSavedSchemaHistory(), cdcStateHandler.compressSchemaHistoryForState())) | ||
| : Optional.<AirbyteSchemaHistoryStorage>empty(); | ||
| final var publisher = new DebeziumRecordPublisher(debeziumPropertiesManager); | ||
| final var queue = new CapacityReportingBlockingQueue<ChangeEvent<String, String>>(queueSize); | ||
| publisher.start(queue, offsetManager, schemaHistoryManager); | ||
| // handle state machine around pub/sub logic. | ||
| final AutoCloseableIterator<ChangeEventWithMetadata> eventIterator = new DebeziumRecordIterator<>( | ||
| queue, | ||
| targetPosition, | ||
| publisher::hasClosed, | ||
| new DebeziumShutdownProcedure<>(queue, publisher::close, publisher::hasClosed), | ||
| firstRecordWaitTime, | ||
| subsequentRecordWaitTime); | ||
|
|
||
| final Duration syncCheckpointDuration = config.has(SYNC_CHECKPOINT_DURATION_PROPERTY) | ||
| ? Duration.ofSeconds(config.get(SYNC_CHECKPOINT_DURATION_PROPERTY).asLong()) | ||
| : SYNC_CHECKPOINT_DURATION; | ||
| final Long syncCheckpointRecords = config.has(SYNC_CHECKPOINT_RECORDS_PROPERTY) | ||
| ? config.get(SYNC_CHECKPOINT_RECORDS_PROPERTY).asLong() | ||
| : SYNC_CHECKPOINT_RECORDS; | ||
|
|
||
| DebeziumMessageProducer messageProducer = new DebeziumMessageProducer(cdcStateHandler, | ||
| targetPosition, | ||
| eventConverter, | ||
| offsetManager, | ||
| schemaHistoryManager); | ||
|
|
||
| // Usually sourceStateIterator requires airbyteStream as input. For DBZ iterator, stream is not used | ||
| // at all thus we will pass in null. | ||
| SourceStateIterator iterator = | ||
| new SourceStateIterator<>(eventIterator, null, messageProducer, new StateEmitFrequency(syncCheckpointRecords, syncCheckpointDuration)); | ||
| return AutoCloseableIterators.fromIterator(iterator); | ||
| } | ||
|
|
||
| public static boolean isAnyStreamIncrementalSyncMode(final ConfiguredAirbyteCatalog catalog) { | ||
| return catalog.getStreams().stream().map(ConfiguredAirbyteStream::getSyncMode) | ||
| .anyMatch(syncMode -> syncMode == SyncMode.INCREMENTAL); | ||
| } | ||
|
|
||
| } |
55 changes: 1 addition & 54 deletions
55
...dk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/CdcMetadataInjector.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 |
|---|---|---|
| @@ -1,56 +1,3 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.integrations.debezium; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
|
|
||
| /** | ||
| * This interface is used to add metadata to the records fetched from the database. For instance, in | ||
| * Postgres we add the lsn to the records. In MySql we add the file name and position to the | ||
| * records. | ||
| */ | ||
| public interface CdcMetadataInjector<T> { | ||
|
|
||
| /** | ||
| * A debezium record contains multiple pieces. Ref : | ||
| * https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-create-events | ||
| * | ||
| * @param event is the actual record which contains data and would be written to the destination | ||
| * @param source contains the metadata about the record and we need to extract that metadata and add | ||
| * it to the event before writing it to destination | ||
| */ | ||
| void addMetaData(ObjectNode event, JsonNode source); | ||
|
|
||
| // TODO : Remove this - it is deprecated. | ||
| default void addMetaDataToRowsFetchedOutsideDebezium(final ObjectNode record, final String transactionTimestamp, final T metadataToAdd) { | ||
| throw new RuntimeException("Not Supported"); | ||
| } | ||
|
|
||
| default void addMetaDataToRowsFetchedOutsideDebezium(final ObjectNode record) { | ||
| throw new RuntimeException("Not Supported"); | ||
| } | ||
|
|
||
| /** | ||
| * As part of Airbyte record we need to add the namespace (schema name) | ||
| * | ||
| * @param source part of debezium record and contains the metadata about the record. We need to | ||
| * extract namespace out of this metadata and return Ref : | ||
| * https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-create-events | ||
| * @return the stream namespace extracted from the change event source. | ||
| */ | ||
| String namespace(JsonNode source); | ||
|
|
||
| /** | ||
| * As part of Airbyte record we need to add the name (e.g. table name) | ||
| * | ||
| * @param source part of debezium record and contains the metadata about the record. We need to | ||
| * extract namespace out of this metadata and return Ref : | ||
| * https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-create-events | ||
| * @return The stream name extracted from the change event source. | ||
| */ | ||
| String name(JsonNode source); | ||
|
|
||
| } |
Oops, something went wrong.