Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java (Java)

Code before:
                                   final BlockingQueue<ApplicationEvent> applicationEventQueue,
                                   final CompletableEventReaper applicationEventReaper,
                                   final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
                                   final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
                                   final Supplier<RequestManagers> requestManagersSupplier,
                                   final AsyncConsumerMetrics asyncConsumerMetrics) {
        this.log = logContext.logger(ApplicationEventHandler.class);
        this.time = time;
        this.applicationEventQueue = applicationEventQueue;
        this.asyncConsumerMetrics = asyncConsumerMetrics;
        this.networkThread = new ConsumerNetworkThread(logContext,
                time,
                applicationEventQueue,
                applicationEventReaper,
                applicationEventProcessorSupplier,
                networkClientDelegateSupplier,
                requestManagersSupplier,
                asyncConsumerMetrics);

        this.networkThread.start(initializationTimeoutMs);
    }

    /**
     * Add an {@link ApplicationEvent} to the handler and then internally invoke {@link #wakeupNetworkThread()}
     * to alert the network I/O thread that it has something to process.
     *
     * @param event An {@link ApplicationEvent} created by the application thread
     */
    public void add(final ApplicationEvent event) {
        Objects.requireNonNull(event, "ApplicationEvent provided to add must be non-null");
        event.setEnqueuedMs(time.milliseconds());
        // Record the updated queue size before actually adding the event to the queue
        // to avoid race conditions (the background thread is continuously removing from this queue)
        asyncConsumerMetrics.recordApplicationEventQueueSize(applicationEventQueue.size() + 1);
        applicationEventQueue.add(event);
        wakeupNetworkThread();
    }

    /**
     * Wakeup the {@link ConsumerNetworkThread network I/O thread} to pull the next event(s) from the queue.
     */
    public void wakeupNetworkThread() {
        networkThread.wakeup();
    }

    /**
     * Returns the delay for which the application thread can safely wait before it should be responsive
     * to results from the request managers. For example, the subscription state can change when heartbeats
     * are sent, so blocking for longer than the heartbeat interval might mean the application thread is not
     * responsive to changes.
     *

Review comment (suggestion):
I'm not sure why this is being set to null. Since this is the constructor, it should be fine to leave the variable as it is. For example:
```
        networkThread = new ConsumerNetworkThread(logContext,
                time,
                applicationEventQueue,
                applicationEventReaper,
                applicationEventProcessorSupplier,
                networkClientDelegateSupplier,
                requestManagersSupplier,
                asyncConsumerMetrics);

        try {
            networkThread.start(initializationTimeoutMs);
        } catch (Exception e) {
            networkThread.close();
            throw e;
        }
```

Code after:
                                   final BlockingQueue<ApplicationEvent> applicationEventQueue,
                                   final CompletableEventReaper applicationEventReaper,
                                   final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
                                   final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
                                   final Supplier<RequestManagers> requestManagersSupplier,
                                   final AsyncConsumerMetrics asyncConsumerMetrics) {
        this.log = logContext.logger(ApplicationEventHandler.class);
        this.time = time;
        this.applicationEventQueue = applicationEventQueue;
        this.asyncConsumerMetrics = asyncConsumerMetrics;
        ConsumerNetworkThread networkThread = new ConsumerNetworkThread(logContext,
                time,
                applicationEventQueue,
                applicationEventReaper,
                applicationEventProcessorSupplier,
                networkClientDelegateSupplier,
                requestManagersSupplier,
                asyncConsumerMetrics);

        try {
            networkThread.start(initializationTimeoutMs);
        } catch (Exception e) {
            try {
                networkThread.close();
            } finally {
                networkThread = null;
            }
            throw e;
        } finally {
            this.networkThread = networkThread;
        }
    }

    /**
     * Add an {@link ApplicationEvent} to the handler and then internally invoke {@link #wakeupNetworkThread()}
     * to alert the network I/O thread that it has something to process.
     *
     * @param event An {@link ApplicationEvent} created by the application thread
     */
    public void add(final ApplicationEvent event) {
        Objects.requireNonNull(event, "ApplicationEvent provided to add must be non-null");
        event.setEnqueuedMs(time.milliseconds());
        // Record the updated queue size before actually adding the event to the queue
        // to avoid race conditions (the background thread is continuously removing from this queue)
        asyncConsumerMetrics.recordApplicationEventQueueSize(applicationEventQueue.size() + 1);
        applicationEventQueue.add(event);
        wakeupNetworkThread();
    }

    /**
     * Wakeup the {@link ConsumerNetworkThread network I/O thread} to pull the next event(s) from the queue.