You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2026/02/11 19:03:18 UTC
[PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
kirktrue opened a new pull request, #21457:
URL: https://github.com/apache/kafka/pull/21457
Updates the `ClassicKafkaConsumer` and `AsyncKafkaConsumer` to clear out the `SubscriptionState` `endOffsetRequested` flag if the `LIST_OFFSETS` call fails.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2818526568
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
In that method, if `isZeroTimestamp` is set to true, it's a synonym for 'only execute a single pass of the loop?' In the case where `isZeroTimestamp` is false, it it's 'possibly execute the pass multiple times.'
We don't want to clear the `partitionEndOffsetsRequested` until we've finished all the passes of the loop we're going to make. So if `shouldClearPartitionEndOffsets` is true but `isZeroTimestamp` is false, clearing the `partitionEndOffsetsRequested` flag could be premature because there could be enough time on the timer for a second pass of the loop that finds the offsets for a partition that the first pass didn't.
I agree that it's confusing. I've tried a couple of different approaches, but they we're much clearer 😦
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830357536
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +155,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (updatePartitionEndOffsetsFlag)
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
I just realized we're passing a subset of the partitions here (there is a retainAll on this collection above), so not really clearing anything effectively if all partitions succeed? (remainingToSearch empty here)
Probably harmful because this is the successful case and for this we also flip the flag when updating HWM/LSO, but this is still not right. Should we pass here the original collection of `timestampsToSearch` instead? To consider also for the failure cause even though we don't alter the collection there, but maybe better to play safe and flip the flag for the same set of partitions that we got intiially?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802316351
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
Just for clarity, I will add the `verify()` bit for sure.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
+ // different from the new async consumer, that will send the LIST_OFFSETS request in the background
+ // thread on the next background thread poll.
+ consumer.poll(Duration.ofMillis(0));
+ }
+
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
+ // Validate the state of the endOffsetRequested flag. It should still be set before the call to
+ // currentLag(), because the previous LIST_OFFSETS call has not received a response.
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Not requesting the log end offset for " + tp0 + " to compute lag as an outstanding request already exists$"
+ );
Review Comment:
Agreed. Will do.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2806431622
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
The flag is only set and only checked on the `currentLag()` path, so the other paths shouldn't need to worry about it.
It's possible for a user to pass in a zero timeout to `offsetsForTimes()`, et al., but we don't need to clear the flag in those cases.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802301895
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java:
##########
@@ -277,6 +277,28 @@ void updateSubscriptionState(Map<TopicPartition, OffsetFetcherUtils.ListOffsetDa
log.trace("Updating high watermark for partition {} to {}", partition, offset);
subscriptionState.updateHighWatermark(partition, offset);
}
+ } else {
+ if (isolationLevel == IsolationLevel.READ_COMMITTED) {
+ log.warn("Not updating last stable offset for partition {} as it is no longer assigned", partition);
+ } else {
+ log.warn("Not updating high watermark for partition {} as it is no longer assigned", partition);
+ }
+ }
+ }
+ }
+
+ /**
+ * If any of the given partitions are assigned, this will clear the partition's 'end offset requested' flag so
+ * that the next attempt to look up the lag will properly issue another <code>LIST_OFFSETS</code> request. This
+ * is only intended to be called when <code>LIST_OFFSETS</code> fails. Successful <code>LIST_OFFSETS</code> calls
+ * should use {@link #updateSubscriptionState(Map, IsolationLevel)}.
+ *
+ * @param partitions Partitions for which the 'end offset requested' flag should be cleared (if still assigned)
+ */
+ void clearPartitionEndOffsetRequests(Collection<TopicPartition> partitions) {
+ for (final TopicPartition partition : partitions) {
+ if (subscriptionState.maybeClearPartitionEndOffsetRequested(partition)) {
+ log.trace("Clearing partition end offset requested for partition {}", partition);
Review Comment:
It's `TRACE`, so no one will ever see it 😉
j/k I'll change it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2819262248
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
> if shouldClearPartitionEndOffsets is true but isZeroTimestamp is false
This is never true, right? (shouldClear is only used for lag)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2805375075
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
Added calls to `verify()` to check when `requestPartitionEndOffset()` is invoked.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2818489870
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
Log message verification removed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2824974816
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -656,21 +659,33 @@ private void process(final CurrentLagEvent event) {
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);
+ if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null) {
+ // The LIST_OFFSETS lag lookup is serialized, so if there's an inflight request it must
+ // finish before another request can be issued. This serialization mechanism is controlled
+ // by the 'end offset requested' flag in SubscriptionState.
+ if (subscriptions.partitionEndOffsetRequested(topicPartition)) {
+ log.info("Not requesting the log end offset for {} to compute lag as an outstanding request already exists", topicPartition);
+ } else {
+ // 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
+ );
+
+ // The currentLag() API is a "best effort" attempt at calling the LIST_OFFSETS RPC. If it
+ // fails, don't retry the attempt internally, but let the user attempt it again.
Review Comment:
Removed changes from `ApplicationEventProcessor` for separation of fixes.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2801132201
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
this log is printed when there is a call to `subscriptionState.requestPartitionEndOffset`, should we better `verify(subscriptionState).requestPartitionEndOffset(tp)`? (we truly care about the func being called, and these log checks easily get flaky)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2805174124
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -559,7 +565,17 @@ private List<NetworkClientDelegate.UnsentRequest> buildListOffsetsRequests(
Map<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(timestampsToSearch, Optional.of(listOffsetsRequestState));
if (timestampsToSearchByNode.isEmpty()) {
- throw new StaleMetadataException();
+ if (listOffsetsRequestState.shouldRetry) {
+ // This exception is caught in prepareFetchOffsetsRequests() and signals to retry this set
+ // of partitions. Only do that if the caller wants to retry LIST_OFFSETS calls, of course.
+ throw new StaleMetadataException();
+ } else {
+ // If the caller doesn't want to retry LIST_OFFSETS calls, go ahead and clear the
+ // 'end offsets requested' flag since there won't be another chance. Then return an empty
+ // list to signal that there are no requests to be sent.
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(timestampsToSearch.keySet());
+ return List.of();
+ }
Review Comment:
ok, good point agree on the unbounded retries gap, but then, isn't that a problem for calls to beginningOffsets/end/offsetsForTimes too? They will timeout correctly at the API level (even expires), but seems they may leave the `OffsetsRequestManager` in the background retrying? (or is there a mechanism in place that I'm missing?)
If so, then agree on addressing the unbounded retries here, but should we address it consistently with all calls retrying while there is time? (instead of having the `shouldRetry` that splits the path)
The gap seems to be having this `OffsetsRequestManager.fetchOffsets` call that has not deadline.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +162,14 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
ok, we both agree we need it for currentLag/timerExpired. But in the way it's called now it applies to all cases, that's my concern. Isn't this going to clear the flag also in the case where there is time left to retry, and there is a partition that didn't have a known leader?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2813326035
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
ack, makes sense that we cannot consolidate on the time check because it depends on the caller.
But then, shouldn't we consolidate on the `shouldClearPartitionEndOffsets`? Why do we need to check `isZeroTimestamp` here? vs simply `if shouldClearPartitionEndOffsets then clear`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -181,27 +203,43 @@ public void onFailure(RuntimeException e) {
}
} while (timer.notExpired());
+ if (shouldClearPartitionEndOffsets) {
Review Comment:
isn't this going to be always false here? (so unneeded?)
`shouldClearPartitionEndOffsets` is true for currentLag only, so time=0 which always early returns above, 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2821725016
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
This is too twisty I'm afraid. A few points.
* I wonder whether it might be better to call `subscriptions.requestPartitionEndOffset` in this method too. Then you are setting it and clearing it all together.
* The flag `isZeroTimestamp` seems misnamed to me. `isZeroTimeout` surely.
* I think that `isZeroTimestamp` (and the equivalent `timer.timeoutMs() == 0L`) has a couple of effects. First, it clears the partition end offset requests flag when the future completes. Second, it exits the loop midway through the first iteration without polling the client.
* I would change the check `if (timer.timeoutMs() == 0L` to use `isZeroTimeout` too.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2800855155
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java:
##########
@@ -277,6 +277,28 @@ void updateSubscriptionState(Map<TopicPartition, OffsetFetcherUtils.ListOffsetDa
log.trace("Updating high watermark for partition {} to {}", partition, offset);
subscriptionState.updateHighWatermark(partition, offset);
}
+ } else {
+ if (isolationLevel == IsolationLevel.READ_COMMITTED) {
+ log.warn("Not updating last stable offset for partition {} as it is no longer assigned", partition);
+ } else {
+ log.warn("Not updating high watermark for partition {} as it is no longer assigned", partition);
+ }
+ }
+ }
+ }
+
+ /**
+ * If any of the given partitions are assigned, this will clear the partition's 'end offset requested' flag so
+ * that the next attempt to look up the lag will properly issue another <code>LIST_OFFSETS</code> request. This
+ * is only intended to be called when <code>LIST_OFFSETS</code> fails. Successful <code>LIST_OFFSETS</code> calls
+ * should use {@link #updateSubscriptionState(Map, IsolationLevel)}.
+ *
+ * @param partitions Partitions for which the 'end offset requested' flag should be cleared (if still assigned)
+ */
+ void clearPartitionEndOffsetRequests(Collection<TopicPartition> partitions) {
+ for (final TopicPartition partition : partitions) {
+ if (subscriptionState.maybeClearPartitionEndOffsetRequested(partition)) {
+ log.trace("Clearing partition end offset requested for partition {}", partition);
Review Comment:
a bit confusing, dup "partition"?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +162,14 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
here we're clearing the flag for the partitions that didn't get offsets yet. I agree we need this if we don't have any time left to retry. But if there's still time, the do-while will try again. In that case, do we want to clear the flag here?
I would imagine we don't, because we'll continue retrying while there is time. It could be the case of missing leader info for instance: we want to keep the flag on for those partitions, hit the `client.awaitMetadataUpdate(timer)` below, and try again in the next iteration of the do-while, right?
If so, I imagine we could take the timer into consideration here? (clear the flag for the failed partitions only if timer expired?). Thoughts?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +162,14 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
}
}
@Override
public void onFailure(RuntimeException e) {
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
same as above
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -659,14 +659,25 @@ public synchronized Long partitionEndOffset(TopicPartition tp, IsolationLevel is
public synchronized void requestPartitionEndOffset(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
- topicPartitionState.requestEndOffset();
+ topicPartitionState.setRequestEndOffset(true);
}
public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
return topicPartitionState.endOffsetRequested();
}
+ public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
+ TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+
+ if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
+ topicPartitionState.setRequestEndOffset(false);
+ return true;
+ } else {
+ return false;
Review Comment:
this return type seems to only be used to log if we performed the action or not. Should we move the log here instead? (simplify this func and the caller)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802291014
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -559,7 +565,17 @@ private List<NetworkClientDelegate.UnsentRequest> buildListOffsetsRequests(
Map<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(timestampsToSearch, Optional.of(listOffsetsRequestState));
if (timestampsToSearchByNode.isEmpty()) {
- throw new StaleMetadataException();
+ if (listOffsetsRequestState.shouldRetry) {
+ // This exception is caught in prepareFetchOffsetsRequests() and signals to retry this set
+ // of partitions. Only do that if the caller wants to retry LIST_OFFSETS calls, of course.
+ throw new StaleMetadataException();
+ } else {
+ // If the caller doesn't want to retry LIST_OFFSETS calls, go ahead and clear the
+ // 'end offsets requested' flag since there won't be another chance. Then return an empty
+ // list to signal that there are no requests to be sent.
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(timestampsToSearch.keySet());
+ return List.of();
+ }
Review Comment:
> In this case we want to refresh metadata and try again once we know the leader. Is there a reason why we want to change that in relationship with the issue of the flag not being cleared?
The existing behavior from `ClassicKafkaConsumer.currentLag()` is to make a single request to `LIST_OFFSETS`. I'm trying to mimic with the `AsyncKafkaConsumer` implementation.
I added a test to `OffsetsRequestManager` that approximates the call to `currentLag()` that will fail due to a missing leader. From looking at the code, it seems as though the `OffsetsRequestManager` will detect the missing leader and retry the `LIST_OFFSETS` repeatedly, without a timeout or limit.
I'm not 100% convinced that the fire-and-forget style approach of `ClassicKafkaConsumer` is correct or optimal either, but it's the approach I'm trying to copy.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2805375938
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
+ // different from the new async consumer, that will send the LIST_OFFSETS request in the background
+ // thread on the next background thread poll.
+ consumer.poll(Duration.ofMillis(0));
+ }
+
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
+ // Validate the state of the endOffsetRequested flag. It should still be set before the call to
+ // currentLag(), because the previous LIST_OFFSETS call has not received a response.
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Not requesting the log end offset for " + tp0 + " to compute lag as an outstanding request already exists$"
+ );
Review Comment:
Added calls to `verify()` to that `requestPartitionEndOffset()` is not invoked in this case.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2805378711
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
FYI: I had to wrap the `SubscriptionState` instance in `spy()` in order to call `verify()` on it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2806625149
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -559,7 +565,17 @@ private List<NetworkClientDelegate.UnsentRequest> buildListOffsetsRequests(
Map<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(timestampsToSearch, Optional.of(listOffsetsRequestState));
if (timestampsToSearchByNode.isEmpty()) {
- throw new StaleMetadataException();
+ if (listOffsetsRequestState.shouldRetry) {
+ // This exception is caught in prepareFetchOffsetsRequests() and signals to retry this set
+ // of partitions. Only do that if the caller wants to retry LIST_OFFSETS calls, of course.
+ throw new StaleMetadataException();
+ } else {
+ // If the caller doesn't want to retry LIST_OFFSETS calls, go ahead and clear the
+ // 'end offsets requested' flag since there won't be another chance. Then return an empty
+ // list to signal that there are no requests to be sent.
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(timestampsToSearch.keySet());
+ return List.of();
+ }
Review Comment:
I've created KAFKA-20188 to track the work for adding a timeout to `OffsetsRequestManager.fetchOffsets()`. So we can focus on that in a separate PR.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830691810
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +155,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (updatePartitionEndOffsetsFlag)
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
There are two steps to clear the flags for all the partitions. In the `onSuccess()` handler, the original set of partitions is split into subsets in `fetchedOffsets` and `partitionsToRetry`:
- The successful partitions in `fetchedOffsets` have their flag cleared via `OffsetFetcherUtils.updateSubscriptionState()`
- The failed partitions in `partitionsToRetry` have their flag cleared via `OffsetFetcherUtils.clearPartitionEndOffsetRequests()`
So by means of one of the of the By the end of `onSuccess()` all partitions will have their respective flags cleared.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830716476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +155,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (updatePartitionEndOffsetsFlag)
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
makes sense, got tricked with the fact that even this is the successful path we're not handling the success cleanup here (only the fails and partial fails) all good, thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "markusdap (via GitHub)" <gi...@apache.org>.
markusdap commented on PR #21457:
URL: https://github.com/apache/kafka/pull/21457#issuecomment-4243590366
@kirktrue Not sure if this is the right place for this, but after updating one of our services to kafka-client 4.1.2 (spring boot 4.0.5), our logs started to get spammed (a few log lines per minute roughly) with the newly added warning log of "Not updating high watermark for partition .... as it is no longer assigned". After some hours investigation I do not see any obvious fix for this (apart from changing log levels for OffsetFetcherUtils class). What is the point of this log? And what can we do to not have this log show up?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2801158046
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
+ // different from the new async consumer, that will send the LIST_OFFSETS request in the background
+ // thread on the next background thread poll.
+ consumer.poll(Duration.ofMillis(0));
+ }
+
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
+ // Validate the state of the endOffsetRequested flag. It should still be set before the call to
+ // currentLag(), because the previous LIST_OFFSETS call has not received a response.
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Not requesting the log end offset for " + tp0 + " to compute lag as an outstanding request already exists$"
+ );
Review Comment:
similar, would it be better to `verify` that we never called `subscriptionState.requestPartitionEndOffset(tp0)`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
+ // different from the new async consumer, that will send the LIST_OFFSETS request in the background
+ // thread on the next background thread poll.
+ consumer.poll(Duration.ofMillis(0));
+ }
+
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
+ // Validate the state of the endOffsetRequested flag. It should still be set before the call to
+ // currentLag(), because the previous LIST_OFFSETS call has not received a response.
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Not requesting the log end offset for " + tp0 + " to compute lag as an outstanding request already exists$"
+ );
Review Comment:
similar, would it be better to `verify` that we `never` called `subscriptionState.requestPartitionEndOffset(tp0)`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802300415
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -656,21 +659,33 @@ private void process(final CurrentLagEvent event) {
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);
+ if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null) {
+ // The LIST_OFFSETS lag lookup is serialized, so if there's an inflight request it must
+ // finish before another request can be issued. This serialization mechanism is controlled
+ // by the 'end offset requested' flag in SubscriptionState.
+ if (subscriptions.partitionEndOffsetRequested(topicPartition)) {
+ log.info("Not requesting the log end offset for {} to compute lag as an outstanding request already exists", topicPartition);
+ } else {
+ // 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
+ );
+
+ // The currentLag() API is a "best effort" attempt at calling the LIST_OFFSETS RPC. If it
+ // fails, don't retry the attempt internally, but let the user attempt it again.
Review Comment:
The leader rediscovery comes from the call to `Metadata.requestUpdate()` that's made in `OffsetsRequestManager.groupListOffsetRequests()`, right? I didn't see that throwing the `StaleMetadataException` did anything besides re-enque the `ListOffsetsRequestState`, which would then send a follow up `LIST_OFFSETS` on the next call to `OffsetsRequestManager.poll()`, right?
My apologies if I'm missing something obvious 😄
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830633662
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -136,11 +137,16 @@ public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartitio
private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestampsToSearch,
Timer timer,
- boolean requireTimestamps) {
+ boolean requireTimestamps,
+ boolean shouldUpdatePartitionEndOffsets) {
Review Comment:
I changed the variable name to `updatePartitionEndOffsetsFlag`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2818526568
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
In that method, `isZeroTimestamp` is kind of a synonym for 'only execute a single pass of the loop?' In the case where `isZeroTimestamp` is true, it it's 'possibly execute the pass multiple times.'
We don't want to clear the `partitionEndOffsetsRequested` until we've finished all the passes of the loop we're going to make. So if `shouldClearPartitionEndOffsets` is true but `isZeroTimestamp` is false, clearing the `partitionEndOffsetsRequested` flag could be premature because there could be enough time on the timer for a second pass of the loop that finds the offsets for a partition that the first pass didn't.
I agree that it's confusing. I've tried a couple of different approaches, but they we're much clearer 😦
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2824977448
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -659,14 +659,25 @@ public synchronized Long partitionEndOffset(TopicPartition tp, IsolationLevel is
public synchronized void requestPartitionEndOffset(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
- topicPartitionState.requestEndOffset();
+ topicPartitionState.setRequestEndOffset(true);
}
public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
return topicPartitionState.endOffsetRequested();
}
+ public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
+ TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+
+ if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
+ topicPartitionState.setRequestEndOffset(false);
+ return true;
+ } else {
+ return false;
Review Comment:
@lianetm—are you OK with leaving the logging in `OffsetFetcherUtils` or should we move it to `SubscriptionState`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802310618
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -659,14 +659,25 @@ public synchronized Long partitionEndOffset(TopicPartition tp, IsolationLevel is
public synchronized void requestPartitionEndOffset(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
- topicPartitionState.requestEndOffset();
+ topicPartitionState.setRequestEndOffset(true);
}
public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
return topicPartitionState.endOffsetRequested();
}
+ public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
+ TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+
+ if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
+ topicPartitionState.setRequestEndOffset(false);
+ return true;
+ } else {
+ return false;
Review Comment:
I left the logging at the higher level for clearing the flag when we fail to find the offset to match where the logging is done when we _do_ find the offset, both in `OffsetFetcherUtils`. That's mostly so that we can enable the log level for a single class to log both the positive and negative cases.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802311464
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -659,14 +659,25 @@ public synchronized Long partitionEndOffset(TopicPartition tp, IsolationLevel is
public synchronized void requestPartitionEndOffset(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
- topicPartitionState.requestEndOffset();
+ topicPartitionState.setRequestEndOffset(true);
}
public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
return topicPartitionState.endOffsetRequested();
}
+ public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
+ TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+
+ if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
+ topicPartitionState.setRequestEndOffset(false);
+ return true;
+ } else {
+ return false;
Review Comment:
But I agree that it's a little wonky.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2800175968
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -656,21 +659,33 @@ private void process(final CurrentLagEvent event) {
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);
+ if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null) {
+ // The LIST_OFFSETS lag lookup is serialized, so if there's an inflight request it must
+ // finish before another request can be issued. This serialization mechanism is controlled
+ // by the 'end offset requested' flag in SubscriptionState.
+ if (subscriptions.partitionEndOffsetRequested(topicPartition)) {
+ log.info("Not requesting the log end offset for {} to compute lag as an outstanding request already exists", topicPartition);
+ } else {
+ // 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
+ );
+
+ // The currentLag() API is a "best effort" attempt at calling the LIST_OFFSETS RPC. If it
+ // fails, don't retry the attempt internally, but let the user attempt it again.
Review Comment:
here is were I think we may be getting something wrong. Agree we don't want to retry the ListOffset triggered from currentLag if it fails. But we do want to retry discovering a leader in order to be able to send the request (which is what the internal retries I pointed in the comment above do).
So if this retry=false is only used to skip that leader re-discovery retry, I think we should remove it (no param needed probably?)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java:
##########
@@ -559,7 +565,17 @@ private List<NetworkClientDelegate.UnsentRequest> buildListOffsetsRequests(
Map<Node, Map<TopicPartition, ListOffsetsRequestData.ListOffsetsPartition>> timestampsToSearchByNode =
groupListOffsetRequests(timestampsToSearch, Optional.of(listOffsetsRequestState));
if (timestampsToSearchByNode.isEmpty()) {
- throw new StaleMetadataException();
+ if (listOffsetsRequestState.shouldRetry) {
+ // This exception is caught in prepareFetchOffsetsRequests() and signals to retry this set
+ // of partitions. Only do that if the caller wants to retry LIST_OFFSETS calls, of course.
+ throw new StaleMetadataException();
+ } else {
+ // If the caller doesn't want to retry LIST_OFFSETS calls, go ahead and clear the
+ // 'end offsets requested' flag since there won't be another chance. Then return an empty
+ // list to signal that there are no requests to be sent.
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(timestampsToSearch.keySet());
+ return List.of();
+ }
Review Comment:
uhm I would say we shouldn't change this retry logic. This internal retry is for the case where when we don't know the leader for a partition when fetching offsets, so we really couldn't even send the request. In this case we want to refresh metadata and try again once we know the leader. Is there a reason why we want to change that in relationship with the issue of the flag not being cleared?
With this change, I'm afraid that if the leader is not known when we call `currentLag`, it would require a next call to currentLag to issue another request, and a third call to lag to get the result vs the current behaviour, where we would retry here to discover the leader and fetch, so the next call to lag would find the results already.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2806148758
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
shouldn't we clear the flag `if (isZeroTimestamp)`?
I see the `shouldClearPartitionEndOffsets` is passed true only from `currentLag`, but what about a call to consumer.`endOffsets`/`beginningOffsets`/`offsetsForTimes` when called with `Duration.ZERO`? how would the flag get cleared?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2806632169
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +162,14 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
I've added an explicit parameter to 'clear end offsets requests' that only the `ClassicKafkaConsumer.currentLag()` sets to true. This should prevent other callers from clearing the flag, regardless of the timeout setting.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2827234894
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -136,11 +137,16 @@ public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartitio
private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestampsToSearch,
Timer timer,
- boolean requireTimestamps) {
+ boolean requireTimestamps,
+ boolean shouldUpdatePartitionEndOffsets) {
Review Comment:
this name is very confusing because it clearly says it will update end offsets (first thing that comes to mind is an actual change to positions, not a flag).
Would it help if we rename to mention it's to update a flag (maybe updatePartitionEndOffsetsFlag), or at least a description of the param?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -659,14 +659,25 @@ public synchronized Long partitionEndOffset(TopicPartition tp, IsolationLevel is
public synchronized void requestPartitionEndOffset(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
- topicPartitionState.requestEndOffset();
+ topicPartitionState.setRequestEndOffset(true);
}
public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
TopicPartitionState topicPartitionState = assignedState(tp);
return topicPartitionState.endOffsetRequested();
}
+ public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
+ TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+
+ if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
+ topicPartitionState.setRequestEndOffset(false);
+ return true;
+ } else {
+ return false;
Review Comment:
ok to leave it if you think it makes troubleshooting easier
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -185,23 +197,48 @@ public void onFailure(RuntimeException e) {
}
public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions, Timer timer) {
- return beginningOrEndOffset(partitions, ListOffsetsRequest.EARLIEST_TIMESTAMP, timer);
+ return beginningOrEndOffset(partitions, ListOffsetsRequest.EARLIEST_TIMESTAMP, timer, false);
}
public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions, Timer timer) {
- return beginningOrEndOffset(partitions, ListOffsetsRequest.LATEST_TIMESTAMP, timer);
+ return beginningOrEndOffset(partitions, ListOffsetsRequest.LATEST_TIMESTAMP, timer, false);
+ }
+
+ public OptionalLong currentLag(TopicPartition topicPartition) {
+ final Long lag = subscriptions.partitionLag(topicPartition, isolationLevel);
+
+ // if the log end offset is not known and hence cannot return lag and there is
+ // no in-flight list offset requested yet,
+ // 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) {
Review Comment:
aren't we missing the check here to ensure there is no request in-flight?
We should also ensure we have a test for this: 2 consecutive calls to currentLag at this level, first one should generate a request, no response, second call should not generate the request. I would expect such test should be failing now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830634774
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -185,23 +197,48 @@ public void onFailure(RuntimeException e) {
}
public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions, Timer timer) {
- return beginningOrEndOffset(partitions, ListOffsetsRequest.EARLIEST_TIMESTAMP, timer);
+ return beginningOrEndOffset(partitions, ListOffsetsRequest.EARLIEST_TIMESTAMP, timer, false);
}
public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions, Timer timer) {
- return beginningOrEndOffset(partitions, ListOffsetsRequest.LATEST_TIMESTAMP, timer);
+ return beginningOrEndOffset(partitions, ListOffsetsRequest.LATEST_TIMESTAMP, timer, false);
+ }
+
+ public OptionalLong currentLag(TopicPartition topicPartition) {
+ final Long lag = subscriptions.partitionLag(topicPartition, isolationLevel);
+
+ // if the log end offset is not known and hence cannot return lag and there is
+ // no in-flight list offset requested yet,
+ // 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) {
Review Comment:
Added a unit test to catch multiple inflight `LIST_OFFESTS` requests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2830716476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +155,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (updatePartitionEndOffsetsFlag)
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
makes sense, got tricked with the retainAll and the fact that even this is the successful path we're not handling the success cleanup here (only the fails and partial fails) all good, thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #21457:
URL: https://github.com/apache/kafka/pull/21457
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2811570115
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
I tend to agree with @lianetm about log checks in the tests. Checking whether the relevant methods were or were not called is better. It was only necessary to make the internal class `OffsetFetcherUtils` public in order to capture its logs. So, I'd vote for not checking the logs in the tests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2825011620
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +169,17 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+
+ if (isZeroTimestamp && shouldClearPartitionEndOffsets)
Review Comment:
I've refactored the `currentLag()` method in `ClassicKafkaConsumer` and `OffsetFetcher` so that the logic resides in the latter. Now `OffsetFetcher.fetchOffsetsByTimes()` can a) set and clear the partition end offset with much closer locality, and b) revert back to the original logic related to the timeout.
PTAL.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #21457:
URL: https://github.com/apache/kafka/pull/21457#issuecomment-4284957240
> @kirktrue I deleted my previous comment as I thought we had a fix, but turns out we did not. Basically our production logs of one of our services is getting spammed with the new "Not updating high watermark for partition..." log. Does this highlight some underlying issue? What is the recommended fix so we do not get this log anymore?
The error `Not updating high watermark for partition X as it is no longer assigned` should only be printed out if a partition reassignment occurs during the offset lookup. Is it happening consistently? Is everything functioning as expected otherwise?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "markusdap (via GitHub)" <gi...@apache.org>.
markusdap commented on PR #21457:
URL: https://github.com/apache/kafka/pull/21457#issuecomment-4260049242
@kirktrue I deleted my previous comment as I thought we had a fix, but turns out we did not. Basically our production logs of one of our services is getting spammed with the new "Not updating high watermark for partition..." log. Does this highlight some underlying issue? What is the recommended fix so we do not get this log anymore?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802315733
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol) throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class, Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class, Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class, Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+ client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
+ );
Review Comment:
Yes, the `verify()` approach is better. That said, I do feel that ensuring that user-facing log messages show up as we expect them to in order to make logs-based troubleshooting possible.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2805371960
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java:
##########
@@ -277,6 +277,28 @@ void updateSubscriptionState(Map<TopicPartition, OffsetFetcherUtils.ListOffsetDa
log.trace("Updating high watermark for partition {} to {}", partition, offset);
subscriptionState.updateHighWatermark(partition, offset);
}
+ } else {
+ if (isolationLevel == IsolationLevel.READ_COMMITTED) {
+ log.warn("Not updating last stable offset for partition {} as it is no longer assigned", partition);
+ } else {
+ log.warn("Not updating high watermark for partition {} as it is no longer assigned", partition);
+ }
+ }
+ }
+ }
+
+ /**
+ * If any of the given partitions are assigned, this will clear the partition's 'end offset requested' flag so
+ * that the next attempt to look up the lag will properly issue another <code>LIST_OFFSETS</code> request. This
+ * is only intended to be called when <code>LIST_OFFSETS</code> fails. Successful <code>LIST_OFFSETS</code> calls
+ * should use {@link #updateSubscriptionState(Map, IsolationLevel)}.
+ *
+ * @param partitions Partitions for which the 'end offset requested' flag should be cleared (if still assigned)
+ */
+ void clearPartitionEndOffsetRequests(Collection<TopicPartition> partitions) {
+ for (final TopicPartition partition : partitions) {
+ if (subscriptionState.maybeClearPartitionEndOffsetRequested(partition)) {
+ log.trace("Clearing partition end offset requested for partition {}", partition);
Review Comment:
Removed the extra "partitions" in the log message.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: ClassicKafkaConsumer does not clear endOffsetRequested flag on failed LIST_OFFSETS calls [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2818533748
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -181,27 +203,43 @@ public void onFailure(RuntimeException e) {
}
} while (timer.notExpired());
+ if (shouldClearPartitionEndOffsets) {
Review Comment:
You're right. I was trying to keep the logic generalized, but maybe that's only more confusing.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20131: SubscriptionState endOffsetRequested remains permanently set if LIST_OFFSETS call fails [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802305160
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java:
##########
@@ -153,11 +162,14 @@ public void onSuccess(ListOffsetResult value) {
remainingToSearch.keySet().retainAll(value.partitionsToRetry);
offsetFetcherUtils.updateSubscriptionState(value.fetchedOffsets, isolationLevel);
+ offsetFetcherUtils.clearPartitionEndOffsetRequests(remainingToSearch.keySet());
Review Comment:
> I agree we need this if we don't have any time left to retry. But if there's still time, the do-while will try again. In that case, do we want to clear the flag here?
That's precisely what happens in the `currentLag()` case, though. It's always using a timeout of `0`, so there's never a second pass in that loop.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org