Exception enqueue(EventContext eventContext,
                          Function<OptionalLong, OptionalLong> deadlineNsCalculator) {
            lock.lock();
            try {
                if (shuttingDown) {
                    return new RejectedExecutionException("The event queue is shutting down");
                }
                if (interrupted) {
                    return new InterruptedException("The event handler thread is interrupted");
                }
                OptionalLong existingDeadlineNs = OptionalLong.empty();
                if (eventContext.tag != null) {
                    EventContext toRemove =
                        tagToEventContext.put(eventContext.tag, eventContext);
                    if (toRemove != null) {
                        existingDeadlineNs = toRemove.deadlineNs;
                        remove(toRemove);
                        size--;
                    }
                }
                OptionalLong deadlineNs = deadlineNsCalculator.apply(existingDeadlineNs);
                boolean queueWasEmpty = head.isSingleton();
                boolean shouldSignal = false;
                switch (eventContext.insertionType) {
                    case APPEND:
                        head.insertBefore(eventContext);
                        if (queueWasEmpty) {
                            shouldSignal = true;
                        }
                        break;
                    case PREPEND:
                        head.insertAfter(eventContext);
                        if (queueWasEmpty) {
                            shouldSignal = true;
                        }
                        break;
                    case DEFERRED:
                        if (!deadlineNs.isPresent()) {
                            return new RuntimeException(
                                "You must specify a deadline for deferred events.");
                        }
                        break;
                }
                if (deadlineNs.isPresent()) {
                    long insertNs = deadlineNs.getAsLong();
                    long prevStartNs = deadlineMap.isEmpty() ? Long.MAX_VALUE : deadlineMap.firstKey();
                    // If the time in nanoseconds is already taken, take the next one.
                    while (deadlineMap.putIfAbsent(insertNs, eventContext) != null) {
                        insertNs++;
                    }
                    eventContext.deadlineNs = OptionalLong.of(insertNs);
                    // If the new timeout is before all the existing ones, wake up the
                    // timeout thread.
                    if (insertNs <= prevStartNs) {
                        shouldSignal = true;
                    }
                }
                size++;
                if (shouldSignal) {
                    cond.signal();
                }
            } finally {
                lock.unlock();
            }
            return null;
        }
A condition variable for waking up the event handler thread.
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java