Github user StefanRRichter commented on a diff in the pull request: https://github.com/apache/flink/pull/5578#discussion_r170982734 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStoreImpl.java --- @@ -300,6 +291,32 @@ private void deleteDirectory(File directory) throws IOException { } } + /** + * Pruning the useless checkpoints. + */ + private void pruneCheckpoints(long checkpointID, boolean breakTheIteration) { + + Iterator<Map.Entry<Long, TaskStateSnapshot>> entryIterator = + storedTaskStateByCheckpointID.entrySet().iterator(); + + final List<Map.Entry<Long, TaskStateSnapshot>> toRemove = new ArrayList<>(); + + while (entryIterator.hasNext()) { + + Map.Entry<Long, TaskStateSnapshot> snapshotEntry = entryIterator.next(); + long entryCheckpointId = snapshotEntry.getKey(); + + if (entryCheckpointId != checkpointID) { --- End diff -- After a second though, while I think this code is currently correct, the case with breaking looks a bit dangerous. Potentially, if the checkpoint id is not there, this would not stop and prune ongoing checkpoints. I wonder if we should make the `if` a bit more complex, but safer (checking that the breaking case never exceeds the checkpoint id). What do you think?
---