convert CDK db-sources submodules to kotlin

Author: stephane-airbyte

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/AirbyteDebeziumHandler.java

@@ -1,149 +0,0 @@
-/*
- * Copyright (c) 2023 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);
-  }
-
-}

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/java/io/airbyte/cdk/integrations/debezium/CdcMetadataInjector.java

@@ -1,56 +0,0 @@
-/*
- * Copyright (c) 2023 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);
-
-}