public void add(long offset, DeferredEvent event) {
        if (!pending.isEmpty()) {
            long lastKey = pending.lastKey();
            if (offset < lastKey) {
                throw new IllegalArgumentException("There is already a deferred event with " +
                    "offset " + lastKey + ". We should not add one with an offset of " +
                    offset + " which is lower than that.");
            }
        }
        List<DeferredEvent> events = pending.computeIfAbsent(offset, k -> new ArrayList<>());
        events.add(event);
        if (log.isTraceEnabled()) {
            log.trace("Adding deferred event {} at offset {}", event, offset);
        }
    }
Add a new deferred event to be completed by the provided offset.

@param offset        The offset to add the new event at.
@param event         The new event.
server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java