class LogicalReplication {

    public PGReplicationStream createReplicationStream(final Connection pgConnection, final String slotName, final LogSequenceNumber startPosition) throws SQLException {
        return pgConnection.unwrap(PGConnection.class).getReplicationAPI()
                .replicationStream()
                .logical()
                .withStartPosition(startPosition)
                .withSlotName(slotName)
                .withSlotOption("include-xids", true)
                .withSlotOption("skip-empty-xacts", true)
                .start();
    }

}