lianetm commented on code in PR #14406:
URL: https://github.com/apache/kafka/pull/14406#discussion_r1333613388
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java:
##########
@@ -217,29 +98,33 @@ public void run() {
}
} catch (final Throwable t) {
log.error("The background thread failed due to unexpected error",
t);
- throw new RuntimeException(t);
+ throw new KafkaException(t);
} finally {
close();
- log.debug("{} closed", getClass());
+ log.debug("Background thread closed");
}
}
+ void initializeResources() {
+ applicationEventProcessor = applicationEventProcessorSupplier.get();
+ networkClientDelegate = networkClientDelegateSupplier.get();
+ requestManagers = requestManagersSupplier.get();
+ }
+
/**
* Poll and process an {@link ApplicationEvent}. It performs the following
tasks:
* 1. Drains and try to process all the requests in the queue.
* 2. Iterate through the registry, poll, and get the next poll time for
the network poll
* 3. Poll the networkClient to send and retrieve the response.
*/
void runOnce() {
- if (!applicationEventQueue.isEmpty()) {
- LinkedList<ApplicationEvent> res = new LinkedList<>();
- this.applicationEventQueue.drainTo(res);
+ LinkedList<ApplicationEvent> events = new LinkedList<>();
+ applicationEventQueue.drainTo(events);
- for (ApplicationEvent event : res) {
- log.debug("Consuming application event: {}", event);
- Objects.requireNonNull(event);
- applicationEventProcessor.process(event);
- }
+ for (ApplicationEvent event : events) {
+ log.trace("Dequeued event: {}", event);
+ Objects.requireNonNull(event);
Review Comment:
Do we expect to have nulls here even if these come from the BlockingQueue
that does not accept nulls?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
