[source-postgres]: fix getLsn() utils function for read replicas (#45397)

Co-authored-by: Marcos Marx <[email protected]>

Author: assaadhjb

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

@@ -174,6 +174,7 @@ corresponds to that version.
 
 | Version    | Date       | Pull Request                                                | Subject                                                                                                                                                        |
 |:-----------|:-----------|:------------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.47.1     | 2024-09-27 | [\#45397](https://github.com/airbytehq/airbyte/pull/45397)  | Allow logical replication from Postgres 16 read-replicas|
 | 0.47.0     | 2024-09-26 | [\#42030](https://github.com/airbytehq/airbyte/pull/42030)  | minor refactor                                                                                                                                                 |
 | 0.46.1     | 2024-09-20 | [\#45700](https://github.com/airbytehq/airbyte/pull/45700)  | Destinations: Fix bug in parsing jsonschema |
 | 0.46.0     | 2024-09-18 | [\#45432](https://github.com/airbytehq/airbyte/pull/45432)  | upgrade all libraries to latest version |

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

@@ -1 +1 @@
-version=0.47.0
+0.47.1

airbyte-cdk/java/airbyte-cdk/datastore-postgres/src/main/kotlin/io/airbyte/cdk/db/PostgresUtils.kt

@@ -18,7 +18,11 @@ object PostgresUtils {
         val jsonNodes =
             database.bufferedResultSetQuery(
                 { conn: Connection ->
-                    conn.createStatement().executeQuery("select * from pg_current_wal_lsn()")
+                    conn
+                        .createStatement()
+                        .executeQuery(
+                            "SELECT CASE WHEN pg_is_in_recovery() THEN pg_last_wal_receive_lsn() ELSE pg_current_wal_lsn() END AS pg_current_wal_lsn;"
+                        )
                 },
                 { resultSet: ResultSet -> JdbcUtils.defaultSourceOperations.rowToJson(resultSet) }
             )