[source-postgres] Add test for legacy version of postgres

airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceLegacyCtidTest.java

@@ -4,13 +4,28 @@
 
 package io.airbyte.integrations.source.postgres;
 
+import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Order;
 
 @Order(2)
 public class CdcPostgresSourceLegacyCtidTest extends CdcPostgresSourceTest {
 
-  protected static String getServerImageName() {
-    return "debezium/postgres:13-bullseye";
+  @Override
+  protected void setBaseImage() {
+    this.postgresImage = BaseImage.POSTGRES_12;
+  }
+
+  @Override
+  @Disabled("https://github.com/airbytehq/airbyte/issues/35267")
+  public void newTableSnapshotTest() {
+
+  }
+
+  @Override
+  @Disabled("https://github.com/airbytehq/airbyte/issues/35267")
+  public void syncShouldIncrementLSN() {
+
   }
 
 }

airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java

@@ -73,9 +73,16 @@
 @Order(1)
 public class CdcPostgresSourceTest extends CdcSourceTest<PostgresSource, PostgresTestDatabase> {
 
+  protected BaseImage postgresImage;
+
+  protected void setBaseImage() {
+    this.postgresImage = getServerImage();
+  }
+
   @Override
   protected PostgresTestDatabase createTestDatabase() {
-    return PostgresTestDatabase.in(getServerImage(), ContainerModifier.CONF).withReplicationSlot();
+    setBaseImage();
+    return PostgresTestDatabase.in(this.postgresImage, ContainerModifier.CONF).withReplicationSlot();
   }
 
   @Override
@@ -101,6 +108,15 @@ protected void setup() {
     testdb.withPublicationForAllTables();
   }
 
+  // For legacy Postgres we will call advanceLsn() after we retrieved target LSN, so that debezium
+  // would not drop any record.
+  // However, that might cause unexpected state and cause failure in the test. Thus we need to bypass
+  // some check if they are on legacy postgres
+  // versions.
+  private boolean isOnLegacyPostgres() {
+    return postgresImage.majorVersion < 15;
+  }
+
   @Test
   void testDebugMode() {
     final JsonNode invalidDebugConfig = testdb.testConfigBuilder()
@@ -196,7 +212,12 @@ private void assertStateTypes(final List<AirbyteStateMessage> stateMessages, fin
       if (Objects.isNull(sharedState)) {
         sharedState = global.getSharedState();
       } else {
-        assertEquals(sharedState, global.getSharedState());
+        // This validation is only true for versions on or after postgres 15. We execute
+        // EPHEMERAL_HEARTBEAT_CREATE_STATEMENTS for earlier versions of
+        // Postgres. See https://github.com/airbytehq/airbyte/pull/33605 for details.
+        if (!isOnLegacyPostgres()) {
+          assertEquals(sharedState, global.getSharedState());
+        }
       }
       assertEquals(1, global.getStreamStates().size());
       final AirbyteStreamState streamState = global.getStreamStates().get(0);
@@ -324,7 +345,11 @@ public void testTwoStreamSync() throws Exception {
       if (Objects.isNull(sharedState)) {
         sharedState = global.getSharedState();
       } else {
-        assertEquals(sharedState, global.getSharedState());
+        // LSN will be advanced for postgres version before 15. See
+        // https://github.com/airbytehq/airbyte/pull/33605
+        if (!isOnLegacyPostgres()) {
+          assertEquals(sharedState, global.getSharedState());
+        }
       }
 
       if (Objects.isNull(firstStreamInState)) {
@@ -755,7 +780,11 @@ protected void assertLsnPositionForSyncShouldIncrementLSN(final Long lsnPosition
     if (syncNumber == 1) {
       assertEquals(1, lsnPosition2.compareTo(lsnPosition1));
     } else if (syncNumber == 2) {
-      assertEquals(0, lsnPosition2.compareTo(lsnPosition1));
+      // Earlier Postgres version will advance lsn even if there is no sync records. See
+      // https://github.com/airbytehq/airbyte/pull/33605.
+      if (!isOnLegacyPostgres()) {
+        assertEquals(0, lsnPosition2.compareTo(lsnPosition1));
+      }
     } else {
       throw new RuntimeException("Unknown sync number " + syncNumber);
     }
@@ -791,7 +820,9 @@ protected void verifyCheckpointStatesByRecords() throws Exception {
         .toListAndClose(secondBatchIterator);
     assertEquals(recordsToCreate, extractRecordMessages(dataFromSecondBatch).size());
     final List<AirbyteStateMessage> stateMessagesCDC = extractStateMessages(dataFromSecondBatch);
-    assertTrue(stateMessagesCDC.size() > 1, "Generated only the final state.");
+    if (!isOnLegacyPostgres()) {
+      assertTrue(stateMessagesCDC.size() > 1, "Generated only the final state.");
+    }
     assertEquals(stateMessagesCDC.size(), stateMessagesCDC.stream().distinct().count(), "There are duplicated states.");
   }
 
@@ -830,7 +861,9 @@ protected void verifyCheckpointStatesBySeconds() throws Exception {
 
     assertEquals(recordsToCreate, extractRecordMessages(dataFromSecondBatch).size());
     final List<AirbyteStateMessage> stateMessagesCDC = extractStateMessages(dataFromSecondBatch);
-    assertTrue(stateMessagesCDC.size() > 1, "Generated only the final state.");
+    if (!isOnLegacyPostgres()) {
+      assertTrue(stateMessagesCDC.size() > 1, "Generated only the final state.");
+    }
     assertEquals(stateMessagesCDC.size(), stateMessagesCDC.stream().distinct().count(), "There are duplicated states.");
   }
 

airbyte-integrations/connectors/source-postgres/src/testFixtures/java/io/airbyte/integrations/source/postgres/PostgresTestDatabase.java

@@ -21,15 +21,17 @@ public class PostgresTestDatabase extends
 
   public static enum BaseImage {
 
-    POSTGRES_16("postgres:16-bullseye"),
-    POSTGRES_12("postgres:12-bullseye"),
-    POSTGRES_9("postgres:9-alpine"),
-    POSTGRES_SSL_DEV("marcosmarxm/postgres-ssl:dev");
+    POSTGRES_16("postgres:16-bullseye", 16),
+    POSTGRES_12("postgres:12-bullseye", 12),
+    POSTGRES_9("postgres:9-alpine", 9),
+    POSTGRES_SSL_DEV("marcosmarxm/postgres-ssl:dev", 16);
 
-    private final String reference;
+    public final String reference;
+    public final int majorVersion;
 
-    private BaseImage(String reference) {
+    private BaseImage(String reference, int majorVersion) {
       this.reference = reference;
+      this.majorVersion = majorVersion;
     };
 
   }