From 054afd6d122afab5f02821fbf0c8649cae666e48 Mon Sep 17 00:00:00 2001 From: Xiaohan Song Date: Thu, 15 Feb 2024 15:31:18 -0800 Subject: [PATCH] [source-postgres] Add test for legacy version of postgres (#35329) --- .../CdcPostgresSourceLegacyCtidTest.java | 19 +++++++- .../postgres/CdcPostgresSourceTest.java | 45 ++++++++++++++++--- .../source/postgres/PostgresTestDatabase.java | 14 +++--- 3 files changed, 64 insertions(+), 14 deletions(-) diff --git a/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceLegacyCtidTest.java b/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceLegacyCtidTest.java index a6d7ecb4d970..53ae215c4da4 100644 --- a/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceLegacyCtidTest.java +++ b/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() { + } } diff --git a/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java b/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java index 490cd4f3e400..551b81ac22f3 100644 --- a/airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java +++ b/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 { + 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 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 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 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."); } diff --git a/airbyte-integrations/connectors/source-postgres/src/testFixtures/java/io/airbyte/integrations/source/postgres/PostgresTestDatabase.java b/airbyte-integrations/connectors/source-postgres/src/testFixtures/java/io/airbyte/integrations/source/postgres/PostgresTestDatabase.java index 155b649e96a8..a86dbd39c351 100644 --- a/airbyte-integrations/connectors/source-postgres/src/testFixtures/java/io/airbyte/integrations/source/postgres/PostgresTestDatabase.java +++ b/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; }; }