AndrewJSchofield commented on code in PR #23123:
URL: https://github.com/apache/kafka/pull/23123#discussion_r3765136673
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -188,6 +189,13 @@ public NetworkClientDelegate.PollResult poll(final long
currentTimeMs) {
public CompletableFuture<Map<TopicPartition, OffsetAndTimestampInternal>>
fetchOffsets(
Map<TopicPartition, Long> timestampsToSearch,
boolean requireTimestamps) {
+ return fetchOffsets(timestampsToSearch, requireTimestamps, false);
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndTimestampInternal>>
fetchOffsets(
+ Map<TopicPartition, Long> timestampsToSearch,
+ boolean requireTimestamps,
+ boolean oneShot) {
if (timestampsToSearch.isEmpty()) {
return CompletableFuture.completedFuture(Collections.emptyMap());
Review Comment:
I know this is not the changed code, but we prefer `Map.of()` nowadays.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -215,6 +224,52 @@ public CompletableFuture<Map<TopicPartition,
OffsetAndTimestampInternal>> fetchO
result.fetchedOffsets));
}
+ /**
+ * Retrieve the consumer's lag on the given partition, i.e. the number of
records between the consumer's
+ * position and the end of the partition (the high watermark, or the last
stable offset when reading with
+ * {@link IsolationLevel#READ_COMMITTED}).
+ *
+ * <p/>
+ *
+ * If the end offset is not known, this issues a <code>LIST_OFFSETS</code>
request in the background so that
+ * the lag may be available on a subsequent call, and returns an empty
result for now. Only one such request
+ * is allowed in flight per partition at a time; that is tracked by the
'end offset requested' flag in
+ * {@link SubscriptionState}, which is set here and cleared when the
request completes, however it completes.
+ *
+ * @param topicPartition Partition to retrieve the lag for
+ * @param isolationLevel Isolation level the lag should be calculated
against
+ * @return The lag, or empty if the end offset for the partition is not
(yet) known
+ */
+ public OptionalLong currentLag(TopicPartition topicPartition,
IsolationLevel isolationLevel) {
+ final Long lag = subscriptionState.partitionLag(topicPartition,
isolationLevel);
+
+ if (lag == null) {
+ // If the log end offset is unknown and there isn't already an
in-flight list offset
+ // request, issue one with the goal that the lag will be available
the next time the
+ // user calls currentLag().
+ if (subscriptionState.partitionEndOffset(topicPartition,
isolationLevel) == null &&
+
offsetFetcherUtils.maybeSetPartitionEndOffsetRequest(topicPartition)) {
+
+ Map<TopicPartition, Long> timestampToSearch =
Collections.singletonMap(
Review Comment:
`Map.of(_,_)` is preferred nowadays.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3064,6 +3058,12 @@ public void
testCurrentLagPreventsMultipleInFlightRequests(GroupProtocol groupPr
consumer.poll(Duration.ofMillis(0));
}
+ // Since the AsyncConsumer uses a background thread, add this barrier
here
Review Comment:
Strictly speaking, adding this wait is just aligning this test with the
others so this comment is a bit out of place. I would remove the comment for
consistency.
--
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]
