You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jy...@apache.org on 2026/09/10 16:47:23 UTC
(kafka) branch 4.3 updated: KAFKA-20187: AsyncKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls (#23423)
This is an automated email from the ASF dual-hosted git repository.
m1a2st pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new d4f5d1ac123 KAFKA-20187: AsyncKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls (#23423)
d4f5d1ac123 is described below
commit d4f5d1ac1237eb687d6059023050a0b2cc36f24b
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Fri Sep 11 00:47:16 2026 +0800
KAFKA-20187: AsyncKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls (#23423)
backport to 4.3
Reviewers: PoAn Yang <pa...@apache.org>
Co-authored-by: Evan Zhou <84...@users.noreply.github.com>
---
.../consumer/internals/OffsetsRequestManager.java | 93 +++++++++++++++++++---
.../events/ApplicationEventProcessor.java | 37 +--------
.../kafka/clients/consumer/KafkaConsumerTest.java | 20 ++---
3 files changed, 95 insertions(+), 55 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
index 23d95811555..2dcf933d4a9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
@@ -44,13 +44,13 @@ import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
@@ -188,15 +188,23 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
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());
+ return CompletableFuture.completedFuture(Map.of());
}
metadata.addTransientTopics(OffsetFetcherUtils.topicsForPartitions(timestampsToSearch.keySet()));
ListOffsetsRequestState listOffsetsRequestState = new ListOffsetsRequestState(
timestampsToSearch,
requireTimestamps,
offsetFetcherUtils,
- isolationLevel);
+ isolationLevel,
+ oneShot);
listOffsetsRequestState.globalResult.whenComplete((result, error) -> {
metadata.clearTransientTopics();
if (error != null) {
@@ -215,6 +223,52 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
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 = Map.of(
+ topicPartition,
+ ListOffsetsRequest.LATEST_TIMESTAMP
+ );
+
+ // The request is issued as 'one shot' so that it always completes rather than being retried
+ // internally on a metadata update. That keeps the 'end offset requested' flag clearing below
+ // reachable on every outcome, so a failed LIST_OFFSETS doesn't block all future lag lookups.
+ // A successful response clears the flag as a side effect of updating the subscription state,
+ // so the call below is a no-op in that case.
+ fetchOffsets(timestampToSearch, false, true).whenComplete((__, error) ->
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(Set.of(topicPartition)));
+ }
+
+ return OptionalLong.empty();
+ }
+
+ return OptionalLong.of(lag);
+ }
+
/**
* Update fetch positions for assigned partitions that do not have a position. This will:
* <ul>
@@ -523,7 +577,10 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
timestampsToSearch, requireTimestamps, listOffsetsRequestState);
requestsToSend.addAll(unsentRequests);
} catch (StaleMetadataException e) {
- requestsToRetry.add(listOffsetsRequestState);
+ if (listOffsetsRequestState.oneShot)
+ listOffsetsRequestState.completeWithResultSoFar();
+ else
+ requestsToRetry.add(listOffsetsRequestState);
}
}
@@ -573,11 +630,8 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
offsetFetcherUtils.updateSubscriptionState(multiNodeResult.fetchedOffsets,
isolationLevel);
- if (listOffsetsRequestState.remainingToSearch.isEmpty()) {
- ListOffsetResult listOffsetResult =
- new ListOffsetResult(listOffsetsRequestState.fetchedOffsets,
- listOffsetsRequestState.remainingToSearch.keySet());
- listOffsetsRequestState.globalResult.complete(listOffsetResult);
+ if (listOffsetsRequestState.remainingToSearch.isEmpty() || listOffsetsRequestState.oneShot) {
+ listOffsetsRequestState.completeWithResultSoFar();
} else {
requestsToRetry.add(listOffsetsRequestState);
metadata.requestUpdate(false);
@@ -826,10 +880,20 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
final OffsetFetcherUtils offsetFetcherUtils;
final IsolationLevel isolationLevel;
+ /**
+ * If true, this request is never held back to be retried on a metadata update. It completes on the
+ * first attempt with whatever offsets were retrieved, leaving it to the caller to decide whether to
+ * issue another request. This is used by
+ * {@link OffsetsRequestManager#currentLag(TopicPartition, IsolationLevel)}, which relies on the
+ * request always completing in order to clear its 'end offset requested' flag.
+ */
+ final boolean oneShot;
+
private ListOffsetsRequestState(Map<TopicPartition, Long> timestampsToSearch,
boolean requireTimestamps,
OffsetFetcherUtils offsetFetcherUtils,
- IsolationLevel isolationLevel) {
+ IsolationLevel isolationLevel,
+ boolean oneShot) {
remainingToSearch = new HashMap<>();
fetchedOffsets = new HashMap<>();
globalResult = new CompletableFuture<>();
@@ -838,12 +902,21 @@ public final class OffsetsRequestManager implements RequestManager, ClusterResou
this.requireTimestamps = requireTimestamps;
this.offsetFetcherUtils = offsetFetcherUtils;
this.isolationLevel = isolationLevel;
+ this.oneShot = oneShot;
}
private void addPartitionsToRetry(Set<TopicPartition> partitionsToRetry) {
remainingToSearch.putAll(partitionsToRetry.stream()
.collect(Collectors.toMap(tp -> tp, timestampsToSearch::get)));
}
+
+ /**
+ * Completes {@link #globalResult} with the offsets retrieved so far, reporting any partitions that
+ * are still outstanding as partitions to retry.
+ */
+ private void completeWithResultSoFar() {
+ globalResult.complete(new ListOffsetResult(fetchedOffsets, remainingToSearch.keySet()));
+ }
}
private static class MultiNodeRequest {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
index d024d3736a4..0bfa0d01ddb 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
@@ -32,22 +32,18 @@ import org.apache.kafka.clients.consumer.internals.ShareMembershipManager;
import org.apache.kafka.clients.consumer.internals.StreamsMembershipManager;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.requests.ListOffsetsRequest;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -666,35 +662,10 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
private void process(final CurrentLagEvent event) {
try {
- final TopicPartition topicPartition = event.partition();
- final IsolationLevel isolationLevel = event.isolationLevel();
- final Long lag = subscriptions.partitionLag(topicPartition, isolationLevel);
-
- final OptionalLong lagOpt;
- if (lag == null) {
- if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null &&
- !subscriptions.partitionEndOffsetRequested(topicPartition)) {
- // 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().
- log.info("Requesting the log end offset for {} in order to compute lag", topicPartition);
- subscriptions.requestPartitionEndOffset(topicPartition);
-
- // Emulates the Consumer.endOffsets() logic...
- Map<TopicPartition, Long> timestampToSearch = Collections.singletonMap(
- topicPartition,
- ListOffsetsRequest.LATEST_TIMESTAMP
- );
-
- requestManagers.offsetsRequestManager.fetchOffsets(timestampToSearch, false);
- }
-
- lagOpt = OptionalLong.empty();
- } else {
- lagOpt = OptionalLong.of(lag);
- }
-
- event.future().complete(lagOpt);
+ event.future().complete(requestManagers.offsetsRequestManager.currentLag(
+ event.partition(),
+ event.isolationLevel()
+ ));
} catch (Exception e) {
event.future().completeExceptionally(e);
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 776b11c86c5..e6cd7494007 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -2784,11 +2784,8 @@ public class KafkaConsumerTest {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test validate that the consumer clears the endOffsetRequested flag, but this is not yet implemented
- // in the CONSUMER group protocol (see KAFKA-20187).
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testCurrentLagPreventsMultipleInFlightRequests(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
@@ -2807,6 +2804,11 @@ public class KafkaConsumerTest {
consumer.poll(Duration.ofMillis(0));
}
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
long count = client.requests().stream()
.filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS))
.count();
@@ -2817,11 +2819,8 @@ public class KafkaConsumerTest {
);
}
- // TODO: this test validate that the consumer clears the endOffsetRequested flag, but this is not yet implemented
- // in the CONSUMER group protocol (see KAFKA-20187).
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
@@ -2875,11 +2874,8 @@ public class KafkaConsumerTest {
);
}
- // TODO: this test validate that the consumer clears the endOffsetRequested flag, but this is not yet implemented
- // in the CONSUMER group protocol (see KAFKA-20187).
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testCurrentLagClearsFlagOnRetriablePartitionError(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);