[DBZ-PGYB] Support for consistent snapshot for an existing slot

[asrinivasanyb]

Summary
This PR is to support consistent snapshot in the case of an existing slot.

In this case, the consistent_point hybrid time is determined from the pg_replication_slots view, specifically from the yb_restart_commit_ht column.

There is an assumption here that this slot has not been used for streaming till this point. If this holds, then the history retention barrier will be in place as of the consistent snapshot time (consistent_point). The snapshot query will be run as of the consistent_point and subsequent streaming will start from the consistent_point of the slot.

Test Plan
Added new test
mvn -Dtest=PostgresConnectorIT#initialSnapshotWithExistingSlot test

[vaibhav-yb]

Just to confirm my understanding after this PR - if a user creates a replication slot and starts the connector with snapshot.mode=never and after streaming if the user restarts the connector in initial mode, he will only get the records snapshotted which would be there if the snapshot was taken in the first place itself.

And in this case, what would happen once the snapshot is completed? Will the records be streamed from the point they were streamed before connector restart?

[asrinivasanyb]

Just to confirm my understanding after this PR - if a user creates a replication slot and starts the connector with snapshot.mode=never and after streaming if the user restarts the connector in initial mode, he will only get the records snapshotted which would be there if the snapshot was taken in the first place itself.

And in this case, what would happen once the snapshot is completed? Will the records be streamed from the point they were streamed before connector restart?

The OffsetState in Kafka would indicate that Streaming has started. So, even when user restarts the connector in INITIAL mode, the streaming will continue from the LSN corresponding to the offsetState stored in Kafka

In the case of the newly added test - initialSnapshotWithExistingSlot - I was relying on the fact that the OffsetState would not be saved in Kafka as I am stopping the connector almost immediately after starting it. In this case, as there is no previous offset in Kafka, a snapshot would be taken and the streaming will start from the consistent_point