m1a2st commented on code in PR #23014:
URL: https://github.com/apache/kafka/pull/23014#discussion_r3776883043
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java:
##########
@@ -69,6 +74,25 @@ protected void maybeThrowAuthFailure(Node node) {
networkClientDelegate.maybeThrowAuthFailure(node);
}
+ /**
+ * {@inheritDoc}
+ *
+ * If any request is in flight, its completion will wake the application
thread regardless of the outcome, so
+ * no separate bound is needed. Otherwise, if any fetchable partition is
still not buffered, it was skipped for a
+ * transient reason, such as reconnect backoff or an unknown leader. Since
nothing else will wake the application
+ * thread in that case, its wait is bounded by {@code retryBackoffMs}.
+ */
+ @Override
+ public long maximumTimeToWait(long currentTimeMs) {
+ if (!nodesWithPendingFetchRequests.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+
+ Set<TopicPartition> buffered = fetchBuffer.bufferedPartitions();
+ boolean hasUnbufferedFetchablePartition =
!subscriptions.fetchablePartitions(tp -> !buffered.contains(tp)).isEmpty();
+ return hasUnbufferedFetchablePartition ? retryBackoffMs :
Long.MAX_VALUE;
Review Comment:
I don't think we need a bound in either case:
* paused: the only thing that makes a paused partition fetchable again is
the application thread itself calling `resume()`. There's no background
activity that needs to wake it up for that. Adding a bound here would just
introduce periodic wakeups that repeatedly find nothing changed.
* pending revocation: this is already covered by following mechanisms:
`collectFetch()`'s wait on `inflightPoll.reconciliationCheckFuture()`, and the
heartbeat manager's `maximumTimeToWait()`, which is already tightly bounded
while a rebalance is in progress.
So I'd keep the `fetchablePartitions().isEmpty()` path unbounded.
`FetchRequestManager` doesn't need to duplicate either of those mechanisms.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
