postgres-source: remove LEGACY state flag (#33437)

airbytehq · Dec 19, 2023 · b3cb243 · b3cb243
1 parent bbec0a1
commit b3cb243
Show file tree

Hide file tree

Showing 19 changed files with 14 additions and 122 deletions.
diff --git a/airbyte-integrations/connectors/source-postgres/build.gradle b/airbyte-integrations/connectors/source-postgres/build.gradle
@@ -13,7 +13,7 @@ java {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.7.1'
+    cdkVersionRequired = '0.7.7'
     features = ['db-sources']
     useLocalCdk = false
 }

diff --git a/airbyte-integrations/connectors/source-postgres/metadata.yaml b/airbyte-integrations/connectors/source-postgres/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.2.27
+  dockerImageTag: 3.3.0
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres

diff --git a/...source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java b/...source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java
@@ -315,13 +315,15 @@ public AirbyteCatalog discover(final JsonNode config) throws Exception {
   @Override
   public JdbcDatabase createDatabase(final JsonNode sourceConfig) throws SQLException {
     final JsonNode jdbcConfig = toDatabaseConfig(sourceConfig);
+    final Map<String, String> connectionProperties = getConnectionProperties(sourceConfig);
     // Create the data source
     final DataSource dataSource = DataSourceFactory.create(
         jdbcConfig.has(JdbcUtils.USERNAME_KEY) ? jdbcConfig.get(JdbcUtils.USERNAME_KEY).asText() : null,
         jdbcConfig.has(JdbcUtils.PASSWORD_KEY) ? jdbcConfig.get(JdbcUtils.PASSWORD_KEY).asText() : null,
-        driverClass,
+        driverClassName,
         jdbcConfig.get(JdbcUtils.JDBC_URL_KEY).asText(),
-        getConnectionProperties(sourceConfig));
+        connectionProperties,
+        getConnectionTimeout(connectionProperties, driverClassName));
     // Record the data source so that it can be closed.
     dataSources.add(dataSource);
 
@@ -688,9 +690,6 @@ protected boolean isNotInternalSchema(final JsonNode jsonNode, final Set<String>
 
   @Override
   protected AirbyteStateType getSupportedStateType(final JsonNode config) {
-    if (!featureFlags.useStreamCapableState()) {
-      return AirbyteStateType.LEGACY;
-    }
     return PostgresUtils.isCdc(config) ? AirbyteStateType.GLOBAL : AirbyteStateType.STREAM;
   }
 

diff --git a/...airbyte/integration_tests/sources/AbstractPostgresSourceSSLCertificateAcceptanceTest.java b/...airbyte/integration_tests/sources/AbstractPostgresSourceSSLCertificateAcceptanceTest.java
@@ -7,8 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.Lists;
 import io.airbyte.cdk.integrations.standardtest.source.TestDestinationEnv;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -34,11 +32,6 @@ public abstract class AbstractPostgresSourceSSLCertificateAcceptanceTest extends
 
   protected PostgresTestDatabase testdb;
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected void setupEnvironment(final TestDestinationEnv environment) throws Exception {
     testdb = PostgresTestDatabase.in(BaseImage.POSTGRES_16, ContainerModifier.CERT)
@@ -105,9 +98,4 @@ protected JsonNode getState() {
     return Jsons.jsonNode(new HashMap<>());
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
 }
diff --git a/...rations/io/airbyte/integration_tests/sources/AbstractSshPostgresSourceAcceptanceTest.java b/...rations/io/airbyte/integration_tests/sources/AbstractSshPostgresSourceAcceptanceTest.java
@@ -13,8 +13,6 @@
 import io.airbyte.cdk.integrations.base.ssh.SshBastionContainer;
 import io.airbyte.cdk.integrations.base.ssh.SshTunnel;
 import io.airbyte.cdk.integrations.standardtest.source.TestDestinationEnv;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.functional.CheckedFunction;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
@@ -77,11 +75,6 @@ private static Database getDatabaseFromConfig(final JsonNode config) {
 
   public abstract SshTunnel.TunnelMethod getTunnelMethod();
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   // todo (cgardens) - dynamically create data by generating a database with a random name instead of
   // requiring data to already be in place.
   @Override
@@ -141,9 +134,4 @@ protected JsonNode getState() {
     return Jsons.jsonNode(new HashMap<>());
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
 }
diff --git a/...ns/io/airbyte/integration_tests/sources/CdcInitialSnapshotPostgresSourceDatatypeTest.java b/...ns/io/airbyte/integration_tests/sources/CdcInitialSnapshotPostgresSourceDatatypeTest.java
@@ -7,8 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.cdk.db.Database;
 import io.airbyte.cdk.integrations.standardtest.source.TestDataHolder;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.ContainerModifier;
@@ -18,11 +16,6 @@ public class CdcInitialSnapshotPostgresSourceDatatypeTest extends AbstractPostgr
 
   private static final String SCHEMA_NAME = "test";
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected Database setupDatabase() throws Exception {
     testdb = PostgresTestDatabase.in(BaseImage.POSTGRES_16, ContainerModifier.CONF)

diff --git a/...te/integrations/io/airbyte/integration_tests/sources/CdcPostgresSourceAcceptanceTest.java b/...te/integrations/io/airbyte/integration_tests/sources/CdcPostgresSourceAcceptanceTest.java
@@ -10,8 +10,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.Lists;
 import io.airbyte.cdk.integrations.standardtest.source.TestDestinationEnv;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -42,11 +40,6 @@ public class CdcPostgresSourceAcceptanceTest extends AbstractPostgresSourceAccep
 
   protected PostgresTestDatabase testdb;
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected void setupEnvironment(final TestDestinationEnv environment) throws Exception {
     testdb = PostgresTestDatabase.in(getServerImage(), ContainerModifier.CONF)

diff --git a/...tegrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java b/...tegrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java
@@ -7,8 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.cdk.db.Database;
 import io.airbyte.cdk.integrations.standardtest.source.TestDataHolder;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -66,11 +64,6 @@ protected void postSetup() throws Exception {
     }
   }
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected Database setupDatabase() {
     testdb = PostgresTestDatabase.in(BaseImage.POSTGRES_16, ContainerModifier.CONF)

diff --git a/...ons/io/airbyte/integration_tests/sources/CloudDeploymentPostgresSourceAcceptanceTest.java b/...ons/io/airbyte/integration_tests/sources/CloudDeploymentPostgresSourceAcceptanceTest.java
@@ -43,9 +43,7 @@ public class CloudDeploymentPostgresSourceAcceptanceTest extends SourceAcceptanc
   @Override
   protected FeatureFlags featureFlags() {
     return FeatureFlagsWrapper.overridingDeploymentMode(
-        FeatureFlagsWrapper.overridingUseStreamCapableState(
-            super.featureFlags(),
-            true),
+        super.featureFlags(),
         AdaptiveSourceRunner.CLOUD_MODE);
   }
 
@@ -123,9 +121,4 @@ protected JsonNode getState() {
     return Jsons.jsonNode(new HashMap<>());
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
 }
diff --git a/...rbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceAcceptanceTest.java b/...rbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceAcceptanceTest.java
@@ -12,8 +12,6 @@
 import io.airbyte.cdk.db.Database;
 import io.airbyte.cdk.db.jdbc.JdbcUtils;
 import io.airbyte.cdk.integrations.standardtest.source.TestDestinationEnv;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -43,11 +41,6 @@ public class PostgresSourceAcceptanceTest extends AbstractPostgresSourceAcceptan
   private PostgresTestDatabase testdb;
   private JsonNode config;
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected void setupEnvironment(final TestDestinationEnv environment) throws Exception {
     testdb = PostgresTestDatabase.in(getServerImage());
@@ -98,11 +91,6 @@ protected JsonNode getState() {
     return Jsons.jsonNode(new HashMap<>());
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
   @Test
   public void testFullRefreshWithRevokingSchemaPermissions() throws Exception {
     prepareEnvForUserWithoutPermissions(testdb.getDatabase());

diff --git a/...airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceDatatypeTest.java b/...airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceDatatypeTest.java
@@ -6,20 +6,13 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.cdk.db.Database;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.ContainerModifier;
 import java.sql.SQLException;
 
 public class PostgresSourceDatatypeTest extends AbstractPostgresSourceDatatypeTest {
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected Database setupDatabase() throws SQLException {
     testdb = PostgresTestDatabase.in(BaseImage.POSTGRES_16, ContainerModifier.CONF)

diff --git a/...e/integrations/io/airbyte/integration_tests/sources/XminPostgresSourceAcceptanceTest.java b/...e/integrations/io/airbyte/integration_tests/sources/XminPostgresSourceAcceptanceTest.java
@@ -7,8 +7,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.Lists;
 import io.airbyte.cdk.integrations.standardtest.source.TestDestinationEnv;
-import io.airbyte.commons.features.FeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -40,11 +38,6 @@ protected JsonNode getConfig() throws Exception {
         .build();
   }
 
-  @Override
-  protected FeatureFlags featureFlags() {
-    return FeatureFlagsWrapper.overridingUseStreamCapableState(super.featureFlags(), true);
-  }
-
   @Override
   protected void setupEnvironment(final TestDestinationEnv environment) throws Exception {
     testdb = PostgresTestDatabase.in(BaseImage.POSTGRES_12)
@@ -100,9 +93,4 @@ protected JsonNode getState() throws Exception {
     return Jsons.jsonNode(new HashMap<>());
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
 }
diff --git a/...postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java b/...postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java
@@ -37,8 +37,6 @@
 import io.airbyte.cdk.integrations.debezium.internals.postgres.PostgresReplicationConnection;
 import io.airbyte.cdk.integrations.util.ConnectorExceptionUtil;
 import io.airbyte.commons.exceptions.ConfigErrorException;
-import io.airbyte.commons.features.EnvVariableFeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.util.AutoCloseableIterator;
 import io.airbyte.commons.util.AutoCloseableIterators;
@@ -82,9 +80,7 @@ protected PostgresTestDatabase createTestDatabase() {
 
   @Override
   protected PostgresSource source() {
-    final var source = new PostgresSource();
-    source.setFeatureFlags(FeatureFlagsWrapper.overridingUseStreamCapableState(new EnvVariableFeatureFlags(), true));
-    return source;
+    return new PostgresSource();
   }
 
   @Override

diff --git a/.../test/java/io/airbyte/integrations/source/postgres/CloudDeploymentPostgresSourceTest.java b/.../test/java/io/airbyte/integrations/source/postgres/CloudDeploymentPostgresSourceTest.java
@@ -67,10 +67,7 @@ private Source source() {
     PostgresSource source = new PostgresSource();
     source.setFeatureFlags(
         FeatureFlagsWrapper.overridingDeploymentMode(
-            FeatureFlagsWrapper.overridingUseStreamCapableState(
-                new EnvVariableFeatureFlags(),
-                true),
-            AdaptiveSourceRunner.CLOUD_MODE));
+            new EnvVariableFeatureFlags(), AdaptiveSourceRunner.CLOUD_MODE));
     return PostgresSource.sshWrappedSource(source);
   }
 

diff --git a/...urce-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresDebugger.java b/...urce-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresDebugger.java
@@ -5,15 +5,12 @@
 package io.airbyte.integrations.source.postgres;
 
 import io.airbyte.cdk.integrations.debug.DebugUtil;
-import io.airbyte.commons.features.EnvVariableFeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 
 public class PostgresDebugger {
 
   @SuppressWarnings({"unchecked", "deprecation", "resource"})
   public static void main(final String[] args) throws Exception {
     final PostgresSource postgresSource = new PostgresSource();
-    postgresSource.setFeatureFlags(FeatureFlagsWrapper.overridingUseStreamCapableState(new EnvVariableFeatureFlags(), true));
     DebugUtil.debug(postgresSource);
   }
 

diff --git a/...c/test/java/io/airbyte/integrations/source/postgres/PostgresJdbcSourceAcceptanceTest.java b/...c/test/java/io/airbyte/integrations/source/postgres/PostgresJdbcSourceAcceptanceTest.java
@@ -20,8 +20,6 @@
 import io.airbyte.cdk.db.jdbc.JdbcUtils;
 import io.airbyte.cdk.integrations.source.jdbc.test.JdbcSourceAcceptanceTest;
 import io.airbyte.cdk.integrations.source.relationaldb.models.DbStreamState;
-import io.airbyte.commons.features.EnvVariableFeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.resources.MoreResources;
 import io.airbyte.commons.util.MoreIterators;
@@ -83,9 +81,7 @@ protected JsonNode config() {
 
   @Override
   protected PostgresSource source() {
-    final var source = new PostgresSource();
-    source.setFeatureFlags(FeatureFlagsWrapper.overridingUseStreamCapableState(new EnvVariableFeatureFlags(), true));
-    return source;
+    return new PostgresSource();
   }
 
   @Override
@@ -367,11 +363,6 @@ void incrementalTimestampCheck() throws Exception {
             getTestMessages().get(2)));
   }
 
-  @Override
-  protected boolean supportsPerStream() {
-    return true;
-  }
-
   /**
    * Postgres Source Error Codes:
    * <p>
@@ -447,7 +438,8 @@ public void testUserHasNoPermissionToDataBase() throws Exception {
   }
 
   @Test
-  void testReadMultipleTablesIncrementally() throws Exception {
+  @Override
+  protected void testReadMultipleTablesIncrementally() throws Exception {
     final var config = config();
     ((ObjectNode) config).put(SYNC_CHECKPOINT_RECORDS_PROPERTY, 1);
     final String namespace = getDefaultNamespace();

diff --git a/...ce-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresSourceTest.java b/...ce-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresSourceTest.java
@@ -27,8 +27,6 @@
 import io.airbyte.cdk.integrations.source.relationaldb.state.StateManager;
 import io.airbyte.cdk.integrations.source.relationaldb.state.StateManagerFactory;
 import io.airbyte.commons.exceptions.ConfigErrorException;
-import io.airbyte.commons.features.EnvVariableFeatureFlags;
-import io.airbyte.commons.features.FeatureFlagsWrapper;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.integrations.source.postgres.PostgresTestDatabase.BaseImage;
@@ -155,9 +153,7 @@ void tearDown() {
   }
 
   public PostgresSource source() {
-    final var source = new PostgresSource();
-    source.setFeatureFlags(FeatureFlagsWrapper.overridingUseStreamCapableState(new EnvVariableFeatureFlags(), true));
-    return source;
+    return new PostgresSource();
   }
 
   private static DSLContext getDslContextWithSpecifiedUser(final JsonNode config, final String username, final String password) {