use common abstraction for CDC via debezium for postgres

airbyte-integrations/connectors/source-debezium/src/testFixtures/java/io/airbyte/integrations/source/debezium/CdcSourceTest.java

@@ -72,13 +72,13 @@ public abstract class CdcSourceTest {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(CdcSourceTest.class);
 
-  private static final String MODELS_SCHEMA = "models_schema";
-  private static final String MODELS_STREAM_NAME = "models";
+  protected static final String MODELS_SCHEMA = "models_schema";
+  protected static final String MODELS_STREAM_NAME = "models";
   private static final Set<String> STREAM_NAMES = Sets
       .newHashSet(MODELS_STREAM_NAME);
-  private static final String COL_ID = "id";
-  private static final String COL_MAKE_ID = "make_id";
-  private static final String COL_MODEL = "model";
+  protected static final String COL_ID = "id";
+  protected static final String COL_MAKE_ID = "make_id";
+  protected static final String COL_MODEL = "model";
   protected static final String DB_NAME = MODELS_SCHEMA;
 
   private static final AirbyteCatalog CATALOG = new AirbyteCatalog().withStreams(List.of(
@@ -90,7 +90,7 @@ public abstract class CdcSourceTest {
           Field.of(COL_MODEL, JsonSchemaPrimitive.STRING))
           .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
           .withSourceDefinedPrimaryKey(List.of(List.of(COL_ID)))));
-  private static final ConfiguredAirbyteCatalog CONFIGURED_CATALOG = CatalogHelpers
+  protected static final ConfiguredAirbyteCatalog CONFIGURED_CATALOG = CatalogHelpers
       .toDefaultConfiguredCatalog(CATALOG);
 
   // set all streams to incremental.
@@ -106,7 +106,7 @@ public abstract class CdcSourceTest {
       Jsons.jsonNode(ImmutableMap.of(COL_ID, 15, COL_MAKE_ID, 2, COL_MODEL, "A 220")),
       Jsons.jsonNode(ImmutableMap.of(COL_ID, 16, COL_MAKE_ID, 2, COL_MODEL, "E 350")));
 
-  protected void setup() {
+  protected void setup() throws SQLException {
     createAndPopulateTables();
   }
 
@@ -156,8 +156,7 @@ private void createAndPopulateActualTable() {
    */
   private void createAndPopulateRandomTable() {
     createSchema(MODELS_SCHEMA + "_random");
-    createTable(MODELS_SCHEMA + "_random",
-        MODELS_STREAM_NAME + "_random",
+    createTable(MODELS_SCHEMA + "_random", MODELS_STREAM_NAME + "_random",
         String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200), PRIMARY KEY (%s)", COL_ID + "_random",
             COL_MAKE_ID + "_random",
             COL_MODEL + "_random", COL_ID + "_random"));
@@ -285,13 +284,13 @@ private void assertExpectedRecords(Set<JsonNode> expectedRecords,
   @Test
   @DisplayName("On the first sync, produce returns records that exist in the database.")
   void testExistingData() throws Exception {
+    CdcTargetPosition targetPosition = cdcLatestTargetPosition();
     final AutoCloseableIterator<AirbyteMessage> read = getSource().read(getConfig(), CONFIGURED_CATALOG, null);
     final List<AirbyteMessage> actualRecords = AutoCloseableIterators.toListAndClose(read);
 
     final Set<AirbyteRecordMessage> recordMessages = extractRecordMessages(actualRecords);
     final List<AirbyteStateMessage> stateMessages = extractStateMessages(actualRecords);
 
-    CdcTargetPosition targetPosition = cdcLatestTargetPosition();
     assertNotNull(targetPosition);
     recordMessages.forEach(record -> {
       assertEquals(extractPosition(record.getData()), targetPosition);
@@ -560,6 +559,17 @@ void testCheck() throws Exception {
 
   @Test
   void testDiscover() throws Exception {
+    final AirbyteCatalog expectedCatalog = expectedCatalogForDiscover();
+    final AirbyteCatalog actualCatalog = getSource().discover(getConfig());
+
+    assertEquals(
+        expectedCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
+            .collect(Collectors.toList()),
+        actualCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
+            .collect(Collectors.toList()));
+  }
+
+  protected AirbyteCatalog expectedCatalogForDiscover() {
     final AirbyteCatalog expectedCatalog = Jsons.clone(CATALOG);
 
     createTable(MODELS_SCHEMA, MODELS_STREAM_NAME + "_2", String.format("%s INTEGER, %s INTEGER, %s VARCHAR(200)", COL_ID, COL_MAKE_ID, COL_MODEL));
@@ -581,14 +591,7 @@ void testDiscover() throws Exception {
 
     streams.add(streamWithoutPK);
     expectedCatalog.withStreams(streams);
-
-    final AirbyteCatalog actualCatalog = getSource().discover(getConfig());
-
-    assertEquals(
-        expectedCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
-            .collect(Collectors.toList()),
-        actualCatalog.getStreams().stream().sorted(Comparator.comparing(AirbyteStream::getName))
-            .collect(Collectors.toList()));
+    return expectedCatalog;
   }
 
   protected abstract CdcTargetPosition cdcLatestTargetPosition();

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -62,7 +62,6 @@ public abstract class AbstractJdbcSource extends AbstractRelationalDbSource<JDBC
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractJdbcSource.class);
 
-  public static final String CDC_LSN = "_ab_cdc_lsn";
   private static final String JDBC_COLUMN_DATABASE_NAME = "TABLE_CAT";
   private static final String JDBC_COLUMN_SCHEMA_NAME = "TABLE_SCHEM";
   private static final String JDBC_COLUMN_TABLE_NAME = "TABLE_NAME";

airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMySqlSourceTest.java

@@ -56,6 +56,7 @@
 import io.airbyte.protocol.models.CatalogHelpers;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.SyncMode;
+import java.sql.SQLException;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -74,7 +75,7 @@ public class CdcMySqlSourceTest extends CdcSourceTest {
   private JsonNode config;
 
   @BeforeEach
-  public void setup() {
+  public void setup() throws SQLException {
     init();
     revokeAllPermissions();
     grantCorrectPermissions();

airbyte-integrations/connectors/source-postgres/build.gradle

@@ -13,6 +13,7 @@ dependencies {
     implementation project(':airbyte-integrations:bases:base-java')
     implementation project(':airbyte-protocol:models')
     implementation project(':airbyte-integrations:connectors:source-jdbc')
+    implementation project(':airbyte-integrations:connectors:source-debezium')
     implementation project(':airbyte-integrations:connectors:source-relational-db')
 
     implementation 'org.apache.commons:commons-lang3:3.11'
@@ -22,6 +23,7 @@ dependencies {
     implementation 'io.debezium:debezium-connector-postgres:1.4.2.Final'
 
     testImplementation testFixtures(project(':airbyte-integrations:connectors:source-jdbc'))
+    testImplementation testFixtures(project(':airbyte-integrations:connectors:source-debezium'))
     testImplementation project(":airbyte-json-validation")
     testImplementation project(':airbyte-test-utils')