divijvaidya commented on code in PR #12590:
URL: https://github.com/apache/kafka/pull/12590#discussion_r1057599467
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -2403,17 +2404,46 @@ 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 {
+ // This is a blocking call bound by the time remaining in
closeTimer
+ 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) {
Review Comment:
AutoCloseable expects a `close()` method without any arguments which means
we cannot pass the timer. However, in the upcoming commit, I introduced a new
utility containing helper functions to reduce boilerplate lambda 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
