tree 048f9723f1d3f661f844bcab47ebd444bd638984
parent 73f761fa73d4200d18f628eef7c79cf91dd1a0bc
author Tzu-Li (Gordon) Tai <tzulitai@apache.org> 1696965499 -0700
committer MartijnVisser <martijn@2symbols.com> 1697008004 +0200

[FLINK-33231] [source] Properly evict offsetsToCommit cache on checkpoint complete if no offsets exist

Prior to this fix, if the offsets to commit for a given checkpoint is empty,
which can be the case if no starting offsets were retrieved from Kafka yet,
then on checkpoint completion the cache is not properly evicted up to the
given checkpoint.

This change fixes this such that in notifyOnCheckpointComplete, we shortcut
the method execution to not need to try to commit the offsets since its
empty anyways, and always remember to evict the cache up to the completed
checkpoint.