kirktrue commented on code in PR #15640:
URL: https://github.com/apache/kafka/pull/15640#discussion_r1579885684
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -144,6 +166,12 @@ void runOnce() {
.map(Optional::get)
.map(rm -> rm.maximumTimeToWait(currentTimeMs))
.reduce(Long.MAX_VALUE, Math::min);
+
+ // "Complete" any events that have expired. This cleanup step should
only be called after the network I/O
+ // thread has made at least one call to poll. This is done to emulate
the behavior of the legacy consumer's
+ // handling of timeouts. The legacy consumer makes at least one
attempt to satisfy any network requests
+ // before checking if a timeout has expired.
Review Comment:
Split into a separate method to accommodate the reworded comment.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -273,9 +301,20 @@ void cleanup() {
log.error("Unexpected error during shutdown. Proceed with
closing.", e);
} finally {
sendUnsentRequests(timer);
+
+ // Copy over the completable events to a separate list, then reap
any incomplete
+ // events on that list.
Review Comment:
Removed.
--
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
