Upon restarting an existing connector that was previously streaming, it would be useful to allow the snapshotter classes to choose whether to resume streaming from the last flushed lsn on a replication slot or the lastCompletelyProcessedLsn on the offset, or start streaming from an lsn for a transaction opened at the beginning of the snapshot phase. This would be helpful for a custom snapshotter that wants to run the snapshot phase but not skip over any changes from the replication slot that occurred while the connector was down.

The snapshot phase updates the starting lsn that the streaming process uses so the message decoder skips any messages with an lsn less than the starting xlog position for the transaction that performs the snapshot. Any writes that happen while the connector is down are skipped. Streaming changes while the connector is down could be added by modifying PostgresSnapshotChangeEventSource#determineSnapshotOffset where we're passing null in for lastCompletelyProcessedLsn to instead pass the lastCompletelyProcessedLsn from the offset or the last flushed lsn on a replication slot based on a flag added to the Snapshotter interface (perhaps by adding a method shouldStreamEventsStartingFromSnapshot).