private void handleEvents() {
            Throwable toDeliver = null;
            EventContext toRun = null;
            boolean wasInterrupted = false;
            while (true) {
                if (toRun != null) {
                    wasInterrupted = toRun.run(log, toDeliver);
                }
                lock.lock();
                try {
                    if (toRun != null) {
                        size--;
                        if (wasInterrupted) {
                            interrupted = wasInterrupted;
                        }
                        toDeliver = null;
                        toRun = null;
                        wasInterrupted = false;
                    }
                    long awaitNs = Long.MAX_VALUE;
                    Map.Entry<Long, EventContext> entry = deadlineMap.firstEntry();
                    if (entry != null) {
                        // Search for timed-out events or deferred events that are ready
                        // to run.
                        long now = time.nanoseconds();
                        long timeoutNs = entry.getKey();
                        EventContext eventContext = entry.getValue();
                        if (timeoutNs <= now) {
                            if (eventContext.insertionType == EventInsertionType.DEFERRED) {
                                // The deferred event is ready to run.  Prepend it to the
                                // queue.  (The value for deferred events is a schedule time
                                // rather than a timeout.)
                                remove(eventContext);
                                toDeliver = null;
                                toRun = eventContext;
                            } else {
                                // not a deferred event, so it is a deadline, and it is timed out.
                                remove(eventContext);
                                toDeliver = new TimeoutException();
                                toRun = eventContext;
                            }
                            continue;
                        } else if (interrupted) {
                            remove(eventContext);
                            toDeliver = new InterruptedException("The event handler thread is interrupted");
                            toRun = eventContext;
                            continue;
                        } else if (shuttingDown) {
                            remove(eventContext);
                            toDeliver = new RejectedExecutionException("The event queue is shutting down");
                            toRun = eventContext;
                            continue;
                        }
                        awaitNs = timeoutNs - now;
                    }
                    if (head.next == head) {
                        if (deadlineMap.isEmpty() && (shuttingDown || interrupted)) {
                            // If there are no more entries to process, and the queue is
                            // closing, exit the thread.
                            return;
                        }
                    } else {
                        if (interrupted) {
                            toDeliver = new InterruptedException("The event handler thread is interrupted");
                        } else {
                            toDeliver = null;
                        }
                        toRun = head.next;
                        remove(toRun);
                        continue;
                    }
                    if (awaitNs == Long.MAX_VALUE) {
                        try {
                            cond.await();
                        } catch (InterruptedException e) {
                            log.warn("Interrupted while waiting for a new event. " +
                                "Shutting down event queue");
                            interrupted = true;
                        }
                    } else {
                        try {
                            cond.awaitNanos(awaitNs);
                        } catch (InterruptedException e) {
                            log.warn("Interrupted while waiting for a deferred event. " +
                                "Shutting down event queue");
                            interrupted = true;
                        }
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
A condition variable for waking up the event handler thread.
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java