[source-postgres] Missing replication slot caused by Azure PG Flex Server failover event leads to data loss

### Connector Name

source-postgres

### Connector Version

3.*

### What step the error happened?

Other

### Revelant information

We're facing some issues/limitations with consuming data from Azure Postgres V13 via CDC with the source-postgres container.

We are faced with an issue while extracting data from our Postgres instance

Source information:
Azure PostgreSQL 13.10
https://learn.microsoft.com/en-us/azure/postgresql/

The problem occurs when an Azure Postgres failover event occurs. The Azure Postgres db occasionally triggers a failover event which causes a new database replica to spin up and the old replica is brought down.

For more information on the HA failover events:
https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/concepts-high-availability

When this occurs, our Airbyte connector fails with a Airbyte error:
"""
Log4j2Appender says: ERROR i.a.i.b.AirbyteExceptionHandler(uncaughtException):26 Something went wrong in the connector. See the logs for more details. java.lang.RuntimeException: Unable establish a connection: Could not connect with provided configuration. Error: Expected exactly one replication slot but found 0. Please read the docs and add a replication slot to your database.
"""

The reason for the above error:
There is a known limitation with the Postgres offering we are consuming data from:
https://learn.microsoft.com/en-us/azure/postgresql/flexible-server/concepts-logical#limitations
The postgres replication slot does not exist on the new replica. The replication slot is destroyed along with the previous DB replica. The new DB replica does not have a replication slot initialised.
https://docs.airbyte.com/integrations/sources/postgres/#configuring-postgres-connector-with-change-data-capture-cdc

Our usual process:
- We need to manually recreated the replication slot on the postgres_db. SQL:
```
select * from pg_create_logical_replication_slot('airbyte_somedb_pgoutput', 'pgoutput');
```
- Resync the job
When we resync the job the LSN offset that we last consumed successfully before the recovery, is at a moment in the past before we re-initialised the connection.
source-postgres then ignores the offset and triggers a full re-sync:
```
Saved offset is before Replication slot's confirmed_flush_lsn, Airbyte will trigger sync from scratch
```

This is an issue for us as we lose some CDC data between the point of failover and the point of recovery:
Yes Airbyte performs a full refresh of the postgres db source, and ingests all of the data that exists at that moment in time (At the time of recovery), but what about the DELETE's that have occurred during the replication slot downtime?

^ Notice that there is a gap between LSN: `1208546759936` - `1210107702704` where we lose out on some CDC data between these windows.

## Conclusion

When this occurs:
```
Saved offset is before Replication slot's confirmed_flush_lsn, Airbyte will trigger sync from scratch
```
We can incur CDC data loss.

## What we want

We would like to re-sync the connector to recover at the exact LSN we last successfully consumed.

## Possible solutions

1. We recover the last LSN we have consumed from the Airbyte state db:
```sql
-- Get state of connection
select state from state where connection_id=""
```

Within the nested object in the state field, the latest LSN we consumed from the WAL is stored in the `lsn_proc` and `lsn_commit` fields:

```json
{"state": {"{\"schema\":null,\"payload\":[\"db_name\",{\"server\":\"db_name\"}]}": "...,\"lsn_proc\":1209989616504,\"lsn_commit\":1209989616504,\"lsn\":1209989616504,..."}}
```

We would like to take this value and re-initialise our sync at the last LSN we have consumed.

This potentially could be achieved by using something along the lines of: [pg_replication_origin_advance](https://www.postgresql.org/docs/current/functions-admin.html#PG-REPLICATION-ORIGIN-ADVANCE) to set our replication to a given LSN.

Interested to hear what folks think about this.

### Relevant log output

```shell
2023-01-01 00:00:00 source > INFO i.a.i.d.i.PostgresDebeziumStateUtil(isSavedOffsetAfterReplicationSlotLSN):58 Replication slot confirmed_flush_lsn : 1209989616504 Saved offset LSN : 1208546759936

2023-01-01 00:00:02 source > WARN i.a.i.s.p.PostgresSource(getIncrementalIterators):381 Saved offset is before Replication slot's confirmed_flush_lsn, Airbyte will trigger sync from scratch

2023-01-01 00:00:03 source > INFO i.a.i.s.p.PostgresCdcTargetPosition(targetPosition):50 identified target lsn: PgLsn{lsn=1210107702704}
```

### Contribute

- [X] Yes, I want to contribute