kirktrue commented on code in PR #14835:
URL: https://github.com/apache/kafka/pull/14835#discussion_r1411470319
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -205,6 +205,26 @@ public void wakeup() {
networkClientDelegate.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.
+ *
+ * @return The maximum delay in milliseconds
+ */
+ public long maximumTimeToWait() {
+ final long currentTimeMs = time.milliseconds();
+ if (requestManagers == null) {
+ return MAX_POLL_TIMEOUT_MS;
+ }
+ return requestManagers.entries().stream()
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(rm -> rm.maximumTimeToWait(currentTimeMs))
+ .reduce(Long.MAX_VALUE, Math::min);
+ }
+
Review Comment:
I'm wondering how "fresh" the "maximum time to wait" value needs to be...
Spitballing here, but... say we introduce a new instance variable in
`ConsumerNetworkThread` that stores a cached value named, say,
`notTotallyFreshMaxWaitCached`? We could then invoke `maximumTimeToWait()` from
inside `runOnce()` and update `notTotallyFreshMaxWaitCached`, like so:
```java
private volatile long notTotallyFreshMaxWaitCached = MAX_POLL_TIMEOUT_MS;
void runOnce() {
// If there are errors processing any events, the error will be
thrown immediately. This will have
// the effect of closing the background thread.
applicationEventProcessor.process();
final long currentTimeMs = time.milliseconds();
final long pollWaitTimeMs = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.poll(currentTimeMs))
.map(networkClientDelegate::addAll)
.reduce(MAX_POLL_TIMEOUT_MS, Math::min);
networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs);
notTotallyFreshMaxWaitCached = requestManagers.entries().stream()
.filter(Optional::isPresent)
.map(Optional::get)
.map(rm -> rm.maximumTimeToWait(currentTimeMs))
.reduce(Long.MAX_VALUE, Math::min);
}
public long maximumTimeToWait() {
if (requestManagers == null) {
return MAX_POLL_TIMEOUT_MS;
}
return notTotallyFreshMaxWaitCached;
}
```
(Or something like that.)
This way we still only touch the `RequestManager`s inside `runOnce()`, so we
don't open overselves up to subtle concurrency issues.
Thoughts?
--
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
