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

Code before:
    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent
    ) {
        this(time, logContext, threadNamePrefix, cleanupEvent, null);
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent,
        Consumer<Long> idleTimeCallback
    ) {
        this.time = time;
        this.cleanupEvent = Objects.requireNonNull(cleanupEvent);
        this.lock = new ReentrantLock();
        this.log = logContext.logger(KafkaEventQueue.class);
        this.eventHandler = new EventHandler();
        this.eventHandlerThread = new KafkaThread(threadNamePrefix + EVENT_HANDLER_THREAD_SUFFIX,
            this.eventHandler, false);
        this.shuttingDown = false;
        this.interrupted = false;
        this.idleTimeCallback = idleTimeCallback != null ? idleTimeCallback : __ -> { };
        this.eventHandlerThread.start();
    }

    public Time time() {
        return time;
    }

    @Override
    public void enqueue(EventInsertionType insertionType,
                        String tag,
                        UnaryOperator<OptionalLong> deadlineNsCalculator,
                        Event event) {
        EventContext eventContext = new EventContext(event, insertionType, tag);
        Exception e = eventHandler.enqueue(eventContext, deadlineNsCalculator);
        if (e != null) {
            eventContext.completeWithException(log, e);
        }
    }

    @Override
    public void cancelDeferred(String tag) {
        eventHandler.cancelDeferred(tag);
    }

    @Override

Review comment (suggestion):
I suggest not using `null` in public interfaces. Please use `Optional<T>` if a parameter is optional.
FYI: https://www.infoq.com/presentations/Null-References-The-Billion-Dollar-Mistake-Tony-Hoare/

Code after:

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent
    ) {
        this(time, logContext, threadNamePrefix, cleanupEvent, __ -> { });
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent,
        Consumer<Long> idleTimeCallback
    ) {
        this.time = time;
        this.cleanupEvent = Objects.requireNonNull(cleanupEvent);
        this.lock = new ReentrantLock();
        this.log = logContext.logger(KafkaEventQueue.class);
        this.eventHandler = new EventHandler();
        this.eventHandlerThread = new KafkaThread(threadNamePrefix + EVENT_HANDLER_THREAD_SUFFIX,
            this.eventHandler, false);
        this.shuttingDown = false;
        this.interrupted = false;
        this.idleTimeCallback = Objects.requireNonNull(idleTimeCallback);
        this.eventHandlerThread.start();
    }

    public Time time() {
        return time;
    }

    @Override
    public void enqueue(EventInsertionType insertionType,
                        String tag,
                        UnaryOperator<OptionalLong> deadlineNsCalculator,
                        Event event) {
        EventContext eventContext = new EventContext(event, insertionType, tag);
        Exception e = eventHandler.enqueue(eventContext, deadlineNsCalculator);
        if (e != null) {
            eventContext.completeWithException(log, e);
        }
    }

    @Override
    public void cancelDeferred(String tag) {
        eventHandler.cancelDeferred(tag);
    }