public void completeUpTo(long offset) {
        Iterator<Entry<Long, List<DeferredEvent>>> iter = pending.entrySet().iterator();
        int numCompleted = 0;
        while (iter.hasNext()) {
            Entry<Long, List<DeferredEvent>> entry = iter.next();
            if (entry.getKey() > offset) {
                break;
            }
            for (DeferredEvent event : entry.getValue()) {
                log.debug("completeUpTo({}): successfully completing {}", offset, event);
                event.complete(null);
                numCompleted++;
            }
            iter.remove();
        }
        if (log.isTraceEnabled()) {
            log.trace("completeUpTo({}): successfully completed {} deferred entries",
                    offset, numCompleted);
        }
    }
Complete some purgatory entries.

@param offset        The offset which the high water mark has advanced to.
server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java