Catch up streaming before snapshot may duplicate messages upon resuming streaming

When using a Postgres connector custom snapshot SPI that performs catch up streaming, duplicated message may be sent during the non-catch up streaming phase.

Normally when a connector gracefully shuts down, the connect framework attempts to commit offsets so the latest committed state gets acked on the replication stream.  While the connector is running, the framework periodically commits offsets. Debezium does not manage triggering an offset commit.  When the catch up streaming phase ends, there may be uncommitted state and the connector is unable to determine when the next commit will occur because the commit timing is externally managed. If a commit is not triggered between the end of the catch up streaming phase and the normal streaming phase after the snapshot, the connector may produce some duplicated messages. 

Although the replication stream may be out of date, the internal OffsetContext is aware of the latest committed offset. When the snapshot phase recreates a new offset after catch up streaming, the previous offset has access to the latest state.