The current mechanism to recover records from the db-history topic is to poll the topic every kafka.recovery.poll.interval.ms until 0 records have been returned for a certain number of attempts kafka.recovery.attempts. This is a problematic approach because if there is some latency (such as heavy load on connect), we may receive 0 records enough times to reach the threshold to break out of the recovery, and therefore never reach the end of the db-history topic. This may build an incomplete internal schema and lead to silent errors.

There is currently no logic to detect if we actually got to the end of the topic. If we didn't, this should probably be a fatal error, since we don't want to continue reading the binlog and potentially skip events (which is the current behavior when the internal schema for the event is missing).

Errors should be logged in this case as well; at the moment this condition can occur on startup undetected with no errors in the log.