Currently the snapshot receiver is buffered, this helps to correctly mark the last record when for example the last table to snapshot is empty. 

Since the same logic is also used by the blocking snapshot it can lead to a strange behavior. 

Suppose that an user signal a blocking snapshot with two tables. The first one completes but the last will go in error. With the current implementation the last record from the first table will not be emitted since it is still in the buffer. 

On the other hand, the initial snapshot has the same sort of problem when the error happens after the snapshot completes and before a new record has been emitted from the streaming. This is why we introduced the notification so that the offset will be forced to be saved.  

Should we revisit the behavior in case of a manageable exception during the blocking snapshot? 

Maybe we can call these 

releaseDataSnapshotLocks(snapshotContext);
for (O offset : offsets) {
    offset.preSnapshotCompletion();
}
snapshotReceiver.completeSnapshot();
for (O offset : offsets) {
    offset.postSnapshotCompletion();
} 

 also in case of exception to leave the blocking snapshot in a clean state.