private void dropSubsumedCheckpoints(long checkpointId) {
        Iterator<Map.Entry<Long, PendingCheckpoint>> entries = pendingCheckpoints.entrySet().iterator();

        while (entries.hasNext()) {
            PendingCheckpoint p = entries.next().getValue();
            // remove all pending checkpoints that are lesser than the current completed checkpoint
            if (p.getCheckpointId() < checkpointId && p.canBeSubsumed()) {
                rememberRecentCheckpointId(p.getCheckpointId());
                failPendingCheckpoint(p, CheckpointFailureReason.CHECKPOINT_SUBSUMED);
                entries.remove();
            }
        }
    }