introduce common abstraction for CDC via debezium

airbyte-integrations/bases/debezium/build.gradle

@@ -0,0 +1,23 @@
+plugins {
+    id "java-test-fixtures"
+}
+
+project.configurations {
+    testFixturesImplementation.extendsFrom implementation
+}
+dependencies {
+    implementation project(':airbyte-protocol:models')
+
+    implementation 'io.debezium:debezium-api:1.4.2.Final'
+    implementation 'io.debezium:debezium-embedded:1.4.2.Final'
+    implementation 'io.debezium:debezium-connector-mysql:1.4.2.Final'
+    implementation 'io.debezium:debezium-connector-postgres:1.4.2.Final'
+
+    testFixturesImplementation project(':airbyte-db')
+    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/src/main/java/io/airbyte/integrations/debezium/AirbyteDebeziumHandler.java

@@ -0,0 +1,137 @@
+/*
+ * 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.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.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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(JsonNode config,
+                                CdcTargetPosition targetPosition,
+                                Properties connectorProperties,
+                                ConfiguredAirbyteCatalog catalog,
+                                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(CdcSavedInfo cdcSavedInfo,
+                                                                             CdcStateHandler cdcStateHandler,
+                                                                             CdcConnectorMetadata cdcConnectorMetadata,
+                                                                             Instant emittedAt) {
+    LOGGER.info("using CDC: {}", true);
+    final AirbyteFileOffsetBackingStore offsetManager = AirbyteFileOffsetBackingStore.initializeState(cdcSavedInfo.getSavedOffset());
+    final AirbyteSchemaHistoryStorage schemaHistoryManager = schemaHistoryManager(cdcSavedInfo);
+    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, cdcConnectorMetadata, 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 = () -> {
+      Map<String, String> offset = offsetManager.read();
+      String dbHistory = trackSchemaHistory ? schemaHistoryManager.read() : null;
+
+      return cdcStateHandler.state(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 AirbyteSchemaHistoryStorage schemaHistoryManager(CdcSavedInfo cdcSavedInfo) {
+    if (trackSchemaHistory) {
+      FilteredFileDatabaseHistory.setDatabaseName(config.get("database").asText());
+      return AirbyteSchemaHistoryStorage.initializeDBHistory(cdcSavedInfo.getSavedSchemaHistory());
+    }
+
+    return null;
+  }
+
+}

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/CdcConnectorMetadata.java

@@ -0,0 +1,41 @@
+/*
+ * 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.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 CdcConnectorMetadata {
+
+  void addMetaData(ObjectNode event, JsonNode source);
+
+  String namespace(JsonNode source);
+
+}

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/CdcSavedInfo.java

@@ -0,0 +1,39 @@
+/*
+ * 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.debezium;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * 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 CdcSavedInfo {
+
+  JsonNode getSavedOffset();
+
+  JsonNode getSavedSchemaHistory();
+
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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.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 state(Map<String, String> offset, String dbHistory);
+
+}

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

@@ -0,0 +1,37 @@
+/*
+ * 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.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
+ */
+public interface CdcTargetPosition {
+
+  boolean reachedTargetPosition(JsonNode valueAsJson);
+
+}