upgrade debezium version for postgres to 1.9.2

airbyte-integrations/bases/debezium-new-version/build.gradle

@@ -0,0 +1,26 @@
+plugins {
+    id "java-test-fixtures"
+}
+
+project.configurations {
+    testFixturesImplementation.extendsFrom implementation
+}
+dependencies {
+    implementation project(':airbyte-protocol:protocol-models')
+    implementation project(':airbyte-db:db-lib')
+
+    implementation 'io.debezium:debezium-api:1.9.2.Final'
+    implementation 'io.debezium:debezium-embedded:1.9.2.Final'
+//    implementation 'io.debezium:debezium-connector-mysql:1.9.2.Final'
+    implementation 'io.debezium:debezium-connector-postgres:1.9.2.Final'
+//    implementation 'io.debezium:debezium-connector-sqlserver:1.9.2.Final'
+    implementation 'org.codehaus.plexus:plexus-utils:3.4.2'
+
+    testFixturesImplementation project(':airbyte-db:db-lib')
+    testFixturesImplementation project(':airbyte-integrations:bases:base-java')
+
+    testFixturesImplementation 'org.junit.jupiter:junit-jupiter-engine:5.4.2'
+    testFixturesImplementation 'org.junit.jupiter:junit-jupiter-api:5.4.2'
+    testFixturesImplementation 'org.junit.jupiter:junit-jupiter-params:5.4.2'
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.util.AutoCloseableIterator;
+import io.airbyte.commons.util.AutoCloseableIterators;
+import io.airbyte.commons.util.CompositeIterator;
+import io.airbyte.commons.util.MoreIterators;
+import io.airbyte.integrations.debezium.internals.AirbyteFileOffsetBackingStore;
+import io.airbyte.integrations.debezium.internals.AirbyteSchemaHistoryStorage;
+import io.airbyte.integrations.debezium.internals.DebeziumEventUtils;
+import io.airbyte.integrations.debezium.internals.DebeziumRecordIterator;
+import io.airbyte.integrations.debezium.internals.DebeziumRecordPublisher;
+import io.airbyte.integrations.debezium.internals.FilteredFileDatabaseHistory;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import io.debezium.engine.ChangeEvent;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Supplier;
+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 {
+
+  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
+   */
+  private static final int QUEUE_CAPACITY = 10000;
+
+  private final Properties connectorProperties;
+  private final JsonNode config;
+  private final CdcTargetPosition targetPosition;
+  private final ConfiguredAirbyteCatalog catalog;
+  private final boolean trackSchemaHistory;
+
+  private final LinkedBlockingQueue<ChangeEvent<String, String>> queue;
+
+  public AirbyteDebeziumHandler(final JsonNode config,
+                                final CdcTargetPosition targetPosition,
+                                final Properties connectorProperties,
+                                final ConfiguredAirbyteCatalog catalog,
+                                final boolean trackSchemaHistory) {
+    this.config = config;
+    this.targetPosition = targetPosition;
+    this.connectorProperties = connectorProperties;
+    this.catalog = catalog;
+    this.trackSchemaHistory = trackSchemaHistory;
+    this.queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
+  }
+
+  public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(final CdcSavedInfoFetcher cdcSavedInfoFetcher,
+                                                                             final CdcStateHandler cdcStateHandler,
+                                                                             final CdcMetadataInjector cdcMetadataInjector,
+                                                                             final Instant emittedAt) {
+    LOGGER.info("using CDC: {}", true);
+    final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeState(cdcSavedInfoFetcher.getSavedOffset());
+    final Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager = schemaHistoryManager(cdcSavedInfoFetcher);
+    final DebeziumRecordPublisher publisher = new DebeziumRecordPublisher(connectorProperties, config, catalog, offsetManager,
+        schemaHistoryManager);
+    publisher.start(queue);
+
+    // handle state machine around pub/sub logic.
+    final AutoCloseableIterator<ChangeEvent<String, String>> eventIterator = new DebeziumRecordIterator(
+        queue,
+        targetPosition,
+        publisher::hasClosed,
+        publisher::close);
+
+    // convert to airbyte message.
+    final AutoCloseableIterator<AirbyteMessage> messageIterator = AutoCloseableIterators
+        .transform(
+            eventIterator,
+            (event) -> DebeziumEventUtils.toAirbyteMessage(event, cdcMetadataInjector, emittedAt));
+
+    // our goal is to get the state at the time this supplier is called (i.e. after all message records
+    // have been produced)
+    final Supplier<AirbyteMessage> stateMessageSupplier = () -> {
+      final Map<String, String> offset = offsetManager.read();
+      final String dbHistory = trackSchemaHistory ? schemaHistoryManager
+          .orElseThrow(() -> new RuntimeException("Schema History Tracking is true but manager is not initialised")).read() : null;
+
+      return cdcStateHandler.saveState(offset, dbHistory);
+    };
+
+    // wrap the supplier in an iterator so that we can concat it to the message iterator.
+    final Iterator<AirbyteMessage> stateMessageIterator = MoreIterators.singletonIteratorFromSupplier(stateMessageSupplier);
+
+    // this structure guarantees that the debezium engine will be closed, before we attempt to emit the
+    // state file. we want this so that we have a guarantee that the debezium offset file (which we use
+    // to produce the state file) is up-to-date.
+    final CompositeIterator<AirbyteMessage> messageIteratorWithStateDecorator =
+        AutoCloseableIterators.concatWithEagerClose(messageIterator, AutoCloseableIterators.fromIterator(stateMessageIterator));
+
+    return Collections.singletonList(messageIteratorWithStateDecorator);
+  }
+
+  private Optional<AirbyteSchemaHistoryStorage> schemaHistoryManager(final CdcSavedInfoFetcher cdcSavedInfoFetcher) {
+    if (trackSchemaHistory) {
+      FilteredFileDatabaseHistory.setDatabaseName(config.get("database").asText());
+      return Optional.of(AirbyteSchemaHistoryStorage.initializeDBHistory(cdcSavedInfoFetcher.getSavedSchemaHistory()));
+    }
+
+    return Optional.empty();
+  }
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/CdcMetadataInjector.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.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 {
+
+  /**
+   * A debezium record contains multiple pieces. Ref :
+   * https://debezium.io/documentation/reference/1.4/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);
+
+  /**
+   * 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.4/connectors/mysql.html#mysql-create-events
+   */
+  String namespace(JsonNode source);
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/CdcSavedInfoFetcher.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Optional;
+
+/**
+ * This interface is used to fetch the saved info required for debezium to run incrementally. Each
+ * connector saves offset and schema history in different manner
+ */
+public interface CdcSavedInfoFetcher {
+
+  JsonNode getSavedOffset();
+
+  Optional<JsonNode> getSavedSchemaHistory();
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/CdcStateHandler.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium;
+
+import io.airbyte.protocol.models.AirbyteMessage;
+import java.util.Map;
+
+/**
+ * This interface is used to allow connectors to save the offset and schema history in the manner
+ * which suits them
+ */
+@FunctionalInterface
+public interface CdcStateHandler {
+
+  AirbyteMessage saveState(Map<String, String> offset, String dbHistory);
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/CdcTargetPosition.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * This interface is used to define the target position at the beginning of the sync so that once we
+ * reach the desired target, we can shutdown the sync. This is needed because it might happen that
+ * while we are syncing the data, new changes are being made in the source database and as a result
+ * we might end up syncing forever. In order to tackle that, we need to define a point to end at the
+ * beginning of the sync
+ */
+public interface CdcTargetPosition {
+
+  boolean reachedTargetPosition(JsonNode valueAsJson);
+
+}

airbyte-integrations/bases/debezium-new-version/src/main/java/io/airbyte/integrations/debezium/internals/AirbyteFileOffsetBackingStore.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium.internals;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.base.Preconditions;
+import io.airbyte.commons.json.Jsons;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.commons.io.FileUtils;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.util.SafeObjectInputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class handles reading and writing a debezium offset file. In many cases it is duplicating
+ * logic in debezium because that logic is not exposed in the public API. We mostly treat the
+ * contents of this state file like a black box. We know it is a Map<ByteBuffer, Bytebuffer>.
+ * We deserialize it to a Map<String, String> so that the state file can be human readable. If
+ * we ever discover that any of the contents of these offset files is not string serializable we
+ * will likely have to drop the human readability support and just base64 encode it.
+ */
+public class AirbyteFileOffsetBackingStore {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AirbyteFileOffsetBackingStore.class);
+
+  private final Path offsetFilePath;
+
+  public AirbyteFileOffsetBackingStore(final Path offsetFilePath) {
+    this.offsetFilePath = offsetFilePath;
+  }
+
+  public Path getOffsetFilePath() {
+    return offsetFilePath;
+  }
+
+  public Map<String, String> read() {
+    final Map<ByteBuffer, ByteBuffer> raw = load();
+
+    return raw.entrySet().stream().collect(Collectors.toMap(
+        e -> byteBufferToString(e.getKey()),
+        e -> byteBufferToString(e.getValue())));
+  }
+
+  @SuppressWarnings("unchecked")
+  public void persist(final JsonNode cdcState) {
+    final Map<String, String> mapAsString =
+        cdcState != null ? Jsons.object(cdcState, Map.class) : Collections.emptyMap();
+    final Map<ByteBuffer, ByteBuffer> mappedAsStrings = mapAsString.entrySet().stream().collect(Collectors.toMap(
+        e -> stringToByteBuffer(e.getKey()),
+        e -> stringToByteBuffer(e.getValue())));
+
+    FileUtils.deleteQuietly(offsetFilePath.toFile());
+    save(mappedAsStrings);
+  }
+
+  private static String byteBufferToString(final ByteBuffer byteBuffer) {
+    Preconditions.checkNotNull(byteBuffer);
+    return new String(byteBuffer.array(), StandardCharsets.UTF_8);
+  }
+
+  private static ByteBuffer stringToByteBuffer(final String s) {
+    Preconditions.checkNotNull(s);
+    return ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8));
+  }
+
+  /**
+   * See FileOffsetBackingStore#load - logic is mostly borrowed from here. duplicated because this
+   * method is not public.
+   */
+  @SuppressWarnings("unchecked")
+  private Map<ByteBuffer, ByteBuffer> load() {
+    try (final SafeObjectInputStream is = new SafeObjectInputStream(Files.newInputStream(offsetFilePath))) {
+      // todo (cgardens) - we currently suppress a security warning for this line. use of readObject from
+      // untrusted sources is considered unsafe. Since the source is controlled by us in this case it
+      // should be safe. That said, changing this implementation to not use readObject would remove some
+      // headache.
+      final Object obj = is.readObject();
+      if (!(obj instanceof HashMap))
+        throw new ConnectException("Expected HashMap but found " + obj.getClass());
+      final Map<byte[], byte[]> raw = (Map<byte[], byte[]>) obj;
+      final Map<ByteBuffer, ByteBuffer> data = new HashMap<>();
+      for (final Map.Entry<byte[], byte[]> mapEntry : raw.entrySet()) {
+        final ByteBuffer key = (mapEntry.getKey() != null) ? ByteBuffer.wrap(mapEntry.getKey()) : null;
+        final ByteBuffer value = (mapEntry.getValue() != null) ? ByteBuffer.wrap(mapEntry.getValue()) : null;
+        data.put(key, value);
+      }
+
+      return data;
+    } catch (final NoSuchFileException | EOFException e) {
+      // NoSuchFileException: Ignore, may be new.
+      // EOFException: Ignore, this means the file was missing or corrupt
+      return Collections.emptyMap();
+    } catch (final IOException | ClassNotFoundException e) {
+      throw new ConnectException(e);
+    }
+  }
+
+  /**
+   * See FileOffsetBackingStore#save - logic is mostly borrowed from here. duplicated because this
+   * method is not public.
+   */
+  private void save(final Map<ByteBuffer, ByteBuffer> data) {
+    try (final ObjectOutputStream os = new ObjectOutputStream(Files.newOutputStream(offsetFilePath))) {
+      final Map<byte[], byte[]> raw = new HashMap<>();
+      for (final Map.Entry<ByteBuffer, ByteBuffer> mapEntry : data.entrySet()) {
+        final byte[] key = (mapEntry.getKey() != null) ? mapEntry.getKey().array() : null;
+        final byte[] value = (mapEntry.getValue() != null) ? mapEntry.getValue().array() : null;
+        raw.put(key, value);
+      }
+      os.writeObject(raw);
+    } catch (final IOException e) {
+      throw new ConnectException(e);
+    }
+  }
+
+  public static AirbyteFileOffsetBackingStore initializeState(final JsonNode cdcState) {
+    final Path cdcWorkingDir;
+    try {
+      cdcWorkingDir = Files.createTempDirectory(Path.of("/tmp"), "cdc-state-offset");
+    } catch (final IOException e) {
+      throw new RuntimeException(e);
+    }
+    final Path cdcOffsetFilePath = cdcWorkingDir.resolve("offset.dat");
+
+    final AirbyteFileOffsetBackingStore offsetManager = new AirbyteFileOffsetBackingStore(cdcOffsetFilePath);
+    offsetManager.persist(cdcState);
+    return offsetManager;
+  }
+
+}