source-postgres: read() throws config exception upon xmin wraparound

airbyte-cdk/java/airbyte-cdk/README.md

@@ -174,6 +174,7 @@ corresponds to that version.
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.36.6  | 2024-06-04 | [\#38875](https://github.com/airbytehq/airbyte/pull/38875) | Add an overridable method to check database condition before reading data                                                                                      |
 | 0.36.5  | 2024-06-01 | [\#38792](https://github.com/airbytehq/airbyte/pull/38792) | Throw config exception if no selectable table exists in user provided schemas                                                                                  |
 | 0.36.4  | 2024-05-31 | [\#38824](https://github.com/airbytehq/airbyte/pull/38824) | Param marked as non-null to nullable in JdbcDestinationHandler for NPE fix                                                                                     |
 | 0.36.2  | 2024-05-29 | [\#38538](https://github.com/airbytehq/airbyte/pull/38357) | Exit connector when encountering a config error.                                                                                                               |

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.36.5
+version=0.36.6

airbyte-cdk/java/airbyte-cdk/db-sources/src/main/kotlin/io/airbyte/cdk/integrations/source/relationaldb/AbstractDbSource.kt

@@ -132,6 +132,8 @@ protected constructor(driverClassName: String) :
 
         val database = createDatabase(config)
 
+        precheckDatabase(database, config)
+
         logPreSyncDebugData(database, catalog)
 
         val fullyQualifiedTableNameToInfo =
@@ -181,6 +183,14 @@ protected constructor(driverClassName: String) :
         }
     }
 
+    // Optional - perform some rechecks before read.
+    // For example, postgres source connector can check if the database has
+    // xmin wraparound
+    @Throws(SQLException::class)
+    protected open fun precheckDatabase(database: Database, config: JsonNode) {
+        /* no-op */
+    }
+
     // Optional - perform any initialization logic before read. For example, source connector
     // can choose to load up state manager here.
     protected open fun initializeForStateManager(

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

@@ -12,9 +12,9 @@ java {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.36.5'
+    cdkVersionRequired = '0.36.6'
     features = ['db-sources', 'datastore-postgres']
-    useLocalCdk = false
+    useLocalCdk = true
 }
 
 application {

airbyte-integrations/connectors/source-postgres/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.4.12
+  dockerImageTag: 3.4.13
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java

@@ -345,6 +345,17 @@ public JdbcDatabase createDatabase(final JsonNode sourceConfig) throws SQLExcept
     return database;
   }
 
+  @Override
+  protected void precheckDatabase(final JdbcDatabase database, final JsonNode config) throws SQLException {
+    if (isXmin(config)) {
+      if (PostgresQueryUtils.getXminStatus(database).getNumWraparound() > 0) {
+        throw new ConfigErrorException("We detected XMIN transaction wraparound in the database, " +
+            "which makes this sync option inefficient and can lead to higher credit consumption. " +
+            "Please change the replication method to CDC or cursor based.");
+      }
+    }
+  }
+
   public static Map<String, String> getConnectionProperties(final JsonNode config) {
     final Map<String, String> customProperties =
         config.has(JdbcUtils.JDBC_URL_PARAMS_KEY)

docs/integrations/sources/postgres.md

@@ -311,6 +311,7 @@ According to Postgres [documentation](https://www.postgresql.org/docs/14/datatyp
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                                    |
 |---------|------------|----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.4.13  | 2024-06-04 | [38875](https://github.com/airbytehq/airbyte/pull/38875) | read() throws config exception upon detecting transaction ID wraparound.                                                                                                    |
 | 3.4.12  | 2024-06-04 | [38836](https://github.com/airbytehq/airbyte/pull/38836) | check() throws config error upon detecting transaction ID wraparound.                                                                                                      |
 | 3.4.11  | 2024-06-04 | [38848](https://github.com/airbytehq/airbyte/pull/38848) | Improve UI message and doc on xmin                                                                                                                                         | 
 | 3.4.10  | 2024-05-29 | [38584](https://github.com/airbytehq/airbyte/pull/38584) | Set is_resumable flag in discover.                                                                                                                                         |