divijvaidya commented on code in PR #12590: URL: https://github.com/apache/kafka/pull/12590#discussion_r1000549642
########## clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java: ########## @@ -2403,17 +2404,44 @@ private ClusterResourceListeners configureClusterResourceListeners(Deserializer< return clusterResourceListeners; } - private void close(long timeoutMs, boolean swallowException) { + private Timer createTimerForRequest(final Duration timeout) { + final Time localTime = (time == null) ? new SystemTime() : time; + return localTime.timer(Math.min(timeout.toMillis(), requestTimeoutMs)); + } + + private void close(Duration timeout, boolean swallowException) { log.trace("Closing the Kafka consumer"); AtomicReference<Throwable> firstException = new AtomicReference<>(); - try { - if (coordinator != null) - coordinator.close(time.timer(Math.min(timeoutMs, requestTimeoutMs))); - } catch (Throwable t) { - firstException.compareAndSet(null, t); - log.error("Failed to close coordinator", t); + + final Timer closeTimer = createTimerForRequest(timeout); + // Close objects with a timeout. The timeout is required because coordinator & fetcher send requests to the + // server in the process of closing which may not respect the overall timeout defined for closing the consumer. + if (coordinator != null) { + try { + coordinator.close(closeTimer); + } catch (Throwable t) { + firstException.compareAndSet(null, t); + log.error("Failed to close consumer coordinator", t); + } } - Utils.closeQuietly(fetcher, "fetcher", firstException); + + if (fetcher != null) { + // the timeout for the session close is at-most the requestTimeoutMs + long remainingDurationInTimeout = Math.max(0, timeout.toMillis() - closeTimer.elapsedMs()); + if (remainingDurationInTimeout > 0) { + remainingDurationInTimeout = Math.min(requestTimeoutMs, remainingDurationInTimeout); + } + + closeTimer.reset(remainingDurationInTimeout); + + try { + fetcher.close(closeTimer); Review Comment: Yes that is correct. The blocking wait is bound by the remaining time in the `closeTimer`. This guarantees that the timeout provided to the close() method here is respected. I will add a comment as FYI in the code. -- 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