Repository: apache/kafka
File: server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java (Java)

Code before:
                        }
                    } else {
                        if (interrupted) {
                            toDeliver = new InterruptedException("The event handler thread is interrupted");
                        } else {
                            toDeliver = null;
                        }
                        toRun = head.next;
                        remove(toRun);
                        continue;
                    }

                    long startIdleNs = time.nanoseconds();
                    try {
                        if (awaitNs == Long.MAX_VALUE) {
                            cond.await();
                        } else {
                            cond.awaitNanos(awaitNs);
                        }
                    } catch (InterruptedException e) {

                        log.warn("Interrupted while waiting for a {} event. " +
                                "Shutting down event queue", (awaitNs == Long.MAX_VALUE) ? "new" : "deferred");
                        interrupted = true;
                    } finally {
                        if (idleTimeCallback != null) {
                            long idleNs = Math.max(time.nanoseconds() - startIdleNs, 0);
                            long idleMs = TimeUnit.NANOSECONDS.toMillis(idleNs);
                            idleTimeCallback.accept(idleMs);
                        }
                    }
                } finally {
                    lock.unlock();
                }
            }
        }

        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);

Review comment (suggestion):
Instead of checking for null, let's have a noop consumer if the user don't specify a consumer callback.

Code after:
                    } else {
                        if (interrupted) {
                            toDeliver = new InterruptedException("The event handler thread is interrupted");
                        } else {
                            toDeliver = null;
                        }
                        toRun = head.next;
                        remove(toRun);
                        continue;
                    }

                    long startIdleMs = time.milliseconds();
                    try {
                        if (awaitNs == Long.MAX_VALUE) {
                            cond.await();
                        } else {
                            cond.awaitNanos(awaitNs);
                        }
                    } catch (InterruptedException e) {
                        log.warn(
                                "Interrupted while waiting for a {} event. Shutting down event queue",
                                (awaitNs == Long.MAX_VALUE) ? "new" : "deferred"
                        );
                        interrupted = true;
                    } finally {
                        idleTimeCallback.accept(Math.max(time.milliseconds() - startIdleMs, 0));
                    }
                } finally {
                    lock.unlock();
                }
            }
        }

        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--;