ableegoldman commented on a change in pull request #11057:
URL: https://github.com/apache/kafka/pull/11057#discussion_r675136826
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -2237,7 +2237,21 @@ public OptionalLong currentLag(TopicPartition
topicPartition) {
acquireAndEnsureOpen();
try {
final Long lag = subscriptions.partitionLag(topicPartition,
isolationLevel);
- return lag == null ? OptionalLong.empty() : OptionalLong.of(lag);
+
+ // if the log end offset is not known and hence cannot return lag,
+ // issue a list offset request for that partition so that next time
+ // we may get the answer; we do not need to wait for the return
value
+ // since we would not try to poll the network client synchronously
+ if (lag == null) {
+ if (subscriptions.partitionEndOffset(topicPartition,
isolationLevel) == null) {
+ log.info("Requesting the log end offset for {} in order to
compute lag", topicPartition);
+ fetcher.endOffsets(Collections.singleton(topicPartition),
time.timer(0L));
Review comment:
Isn't this a blocking call? I couldn't find anything that asserted yes
or no in the javadocs, but this ultimately calls down into
`Fetcher#fetchOffsetsByTimes` which does seem to wait for the request future to
complete (in fact, it seems to be doing a busy wait on the timer...? that
doesn't seem right 🤔 )
--
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
