You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lianetm (via GitHub)" <gi...@apache.org> on 2026/02/12 15:26:37 UTC
[PR] KAFKA-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
lianetm opened a new pull request, #21464:
URL: https://github.com/apache/kafka/pull/21464
Fix to ensure consumer poll/committed do not fail fatally on retriable partition-level errors when fetching committed offsets for a topic that is not known to the GC.
This issue didn't happen before using topic IDs on offset fetch, because when fetching with topic names the GC returns no error for unknown partitions (just includes offset -1). But with topic IDs (OffsetFetch v10+), if a topic is not known to the GC, the OffsetFetch response includes UNKNOWN_TOPIC_ID as error.
The fix is to let poll/committed retry while there is time. If the error does not recover, fetch offsets still returns gracefully (offsets for the partitions that have, null for the partitions that were deleted). This keeps the contract of the committed API, and ensures poll uses the committed offsets available (it will use partition offsets for partitions that still need positions)
Added integration tests that fail without the fix, to show the gaps.
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3924069227
> * as the the UNKNOWN_TOPIC_OR_PARTITION, it's indeed never returned by the broker on the offsetfetch path anymore, but it was returned by older broker (e.g, [3.1](https://github.com/apache/kafka/blob/2b57b38f9306ac73f18d94d55c0158530fd444df/core/src/main/scala/kafka/server/KafkaApis.scala#L1344-L1345) broker when using zookeeper and not finding the topic for the offsets there). So I guess we should keep the handling on the client for backwards compatibility
I wonder why it was dropped starting from v1. Was it missed during a refactoring (e.g. when batching was introduced)?
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21464:
URL: https://github.com/apache/kafka/pull/21464#discussion_r2806543578
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -552,24 +557,99 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +
"outbound buffer: {}", fetchRequest);
}
- if (error == null) {
- maybeUpdateLastSeenEpochIfNewer(res);
- result.complete(res);
- } else {
- if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
- if (fetchRequest.isExpired()) {
- log.debug("OffsetFetch request for {} timed out and won't be retried anymore", fetchRequest.requestedPartitions);
- result.completeExceptionally(maybeWrapAsTimeoutException(error));
- } else {
- fetchRequest.resetFuture();
- fetchOffsetsWithRetries(fetchRequest, result);
- }
- } else
- result.completeExceptionally(error);
+
+ // Success: no group-level error and no partition-level errors
+ if (error == null && !res.hasRetriablePartitionErrors()) {
Review Comment:
nit: it may be just me, but it feels like it may be easier to read if the order of ifs were:
```
if (error != null)
...
if (res.hasRetriablePartitionErrors())
...
// success
...
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -284,8 +284,15 @@ private void process(final FetchCommittedOffsetsEvent event) {
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
- CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
- future.whenComplete(complete(event.future()));
+ CompletableFuture<CommitRequestManager.OffsetFetchResult> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
+ future.whenComplete((result, error) -> {
+ if (error != null) {
+ event.future().completeExceptionally(error);
+ } else {
+ Map<TopicPartition, OffsetAndMetadata> offsetMap = result.toOffsetMapWithNulls();
Review Comment:
I understand that it's okay to return nulls here, as it's documented already: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L1315,
but is it possible now to receive a null without this change?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -1194,13 +1336,17 @@ private void onSuccess(final long currentTimeMs,
future.completeExceptionally(new UnstableOffsetCommitException("There are " +
"unstable offsets for the requested topic partitions"));
} else {
Review Comment:
nit: else if
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3891710799
I have a few follow-ups I will address:
- tests at the request level to validate/show the different responses for deleted topics on OffsetFetch topic names vs topic Ids (opening separate PR)
- review usage of UNKNOWN_TOPIC_OR_PARTITION. The client expect its, but seems to never be sent by the broker. Will confirm and clean up client-side
- review UNSTABLE_OFFSET handling on the client. This is the other retriable partiion-level , and it's handled fatally 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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21464:
URL: https://github.com/apache/kafka/pull/21464#discussion_r2814212517
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -284,8 +284,15 @@ private void process(final FetchCommittedOffsetsEvent event) {
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
- CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
- future.whenComplete(complete(event.future()));
+ CompletableFuture<CommitRequestManager.OffsetFetchResult> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
+ future.whenComplete((result, error) -> {
+ if (error != null) {
+ event.future().completeExceptionally(error);
+ } else {
+ Map<TopicPartition, OffsetAndMetadata> offsetMap = result.toOffsetMapWithNulls();
Review Comment:
yes, without this change we used to get nulls when using older brokers (v<4.2). It's with 4.2 that we need this change to continue getting those nulls as expected (e.g. when topic is not known to the GC)
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3922808213
Since this is a response-level change, it's usually documented in the response definition (already mentions the new error on version 10+)
https://github.com/apache/kafka/blob/434ae67d479f7e91f5bd08d58fde5294f31c7980/clients/src/main/resources/common/message/OffsetFetchResponse.json#L54
I also added another comment at a higher level to make the new error clearer.
To complete the documentation story, from the consumer API level the behaviour remains unchanged (null to the user if topic not found), and also documented in the java docs: https://github.com/apache/kafka/blob/434ae67d479f7e91f5bd08d58fde5294f31c7980/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L1581-L1583
LMK, 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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #21464:
URL: https://github.com/apache/kafka/pull/21464
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3951457566
> I wonder why it was dropped starting from v1. Was it missed during a refactoring (e.g. when batching was introduced)?
From what I see in the history it was indeed dropped when group batching was introduced with the implementation of KIP-709, mainly because with that we split the handling to read from zk vs read offsets from the internal topic (that made the difference)
With the split handling introduced in that PR, v0 reads offsets from zk and throws UNKNOWN_TOPIC_OR_PARTITION if not found. And v1+ reads offsets from the internal topic, and the GC returns no error + offsets -1 if not found. This func introduced for KIP-709 https://github.com/apache/kafka/pull/10962/changes#diff-cc056b4960ededba37a438b1454f0f3c5ff5e8ad5e6d2ec9a08e813ca056ffebR1316
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3922314818
As for the follow-ups:
- here is the PR with the integration tests to validate the different responses depending on topicId/name https://github.com/apache/kafka/pull/21509
- checked UNSTABLE_OFFSET case, it's ok because it just propagates the `UnstableOffsetCommitException` (which is retriable)
- as the the UNKNOWN_TOPIC_OR_PARTITION, it's indeed never returned by the broker on the offsetfetch path anymore, but it was returned by older broker (e.g, [3.1](https://github.com/apache/kafka/blob/2b57b38f9306ac73f18d94d55c0158530fd444df/core/src/main/scala/kafka/server/KafkaApis.scala#L1344-L1345) broker when using zookeeper and not finding the topic for the offsets there). So I guess we should keep the handling on the client for backwards compatibility
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21464:
URL: https://github.com/apache/kafka/pull/21464#discussion_r2818118637
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -284,8 +284,15 @@ private void process(final FetchCommittedOffsetsEvent event) {
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
- CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
- future.whenComplete(complete(event.future()));
+ CompletableFuture<CommitRequestManager.OffsetFetchResult> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
+ future.whenComplete((result, error) -> {
+ if (error != null) {
+ event.future().completeExceptionally(error);
+ } else {
+ Map<TopicPartition, OffsetAndMetadata> offsetMap = result.toOffsetMapWithNulls();
Review Comment:
thank you for the explanation
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3922543601
Thanks for the review @Nikita-Shupletsov! Comments addressed (sorry I had missed some).
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21464:
URL: https://github.com/apache/kafka/pull/21464#discussion_r2800656267
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -552,24 +557,99 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +
"outbound buffer: {}", fetchRequest);
}
- if (error == null) {
- maybeUpdateLastSeenEpochIfNewer(res);
- result.complete(res);
- } else {
- if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
- if (fetchRequest.isExpired()) {
- log.debug("OffsetFetch request for {} timed out and won't be retried anymore", fetchRequest.requestedPartitions);
- result.completeExceptionally(maybeWrapAsTimeoutException(error));
- } else {
- fetchRequest.resetFuture();
- fetchOffsetsWithRetries(fetchRequest, result);
- }
- } else
- result.completeExceptionally(error);
+
+ // Success: no group-level error and no partition-level errors
+ if (error == null && !res.hasRetriablePartitionErrors()) {
+ handleSuccessfulOffsetFetch(result, res);
+ return;
+ }
+
+ // Group-level error
+ if (error != null) {
+ handleGroupLevelError(fetchRequest, result, error);
+ return;
}
+
+ // Partition-level errors
+ handleRetriablePartitionErrors(fetchRequest, result, res);
});
}
+ /**
+ * Handles a successful offset fetch response with no errors.
+ */
+ private void handleSuccessfulOffsetFetch(final CompletableFuture<OffsetFetchResult> result,
+ final OffsetFetchResult res) {
+ maybeUpdateLastSeenEpochIfNewer(res.offsets());
+ result.complete(res);
+ }
+
+ /**
+ * Handles group-level errors from an offset fetch request.
+ * Group-level errors indicate the entire request failed (e.g., coordinator unavailable).
+ */
+ private void handleGroupLevelError(final OffsetFetchRequestState fetchRequest,
+ final CompletableFuture<OffsetFetchResult> result,
+ final Throwable error) {
+ boolean isRetriable = (error instanceof RetriableException) ||
+ isStaleEpochErrorAndValidEpochAvailable(error);
+
+ if (!isRetriable) {
+ result.completeExceptionally(error);
+ return;
+ }
+
+ if (fetchRequest.isExpired()) {
+ log.debug("OffsetFetch request for {} timed out and won't be retried anymore",
+ fetchRequest.requestedPartitions);
+ result.completeExceptionally(maybeWrapAsTimeoutException(error));
+ return;
+ }
+
+ retryOffsetFetchOnError(fetchRequest, result, "retriable error: " + error.getMessage());
+ }
+
+ /**
+ * Handles retriable partition-level errors from an offset fetch response.
+ *
+ * <p>The only retriable partition errors are UNKNOWN_TOPIC_ID and UNKNOWN_TOPIC_OR_PARTITION.
+ * When expired or not enough time for another retry, we return partial results with null for
+ * the errored partitions. We check against {@code remainingBackoffMs} to ensure we complete
+ * before the {@link org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper}
+ * expires the event with TimeoutException.
+ */
+ private void handleRetriablePartitionErrors(final OffsetFetchRequestState fetchRequest,
Review Comment:
this is the functional difference introduced with this PR: retry on retriable partition-level errors (UNKNOWN_TOPIC_ID and UNKNOWN_TOPIC_OR_PARTITION for now), return whatever results we have if there is no more time to retry.
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on PR #21464:
URL: https://github.com/apache/kafka/pull/21464#issuecomment-3923615328
makes sense, thanks a lot for the explanation.
--
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-20165: Handle retriable partition errors gracefully on OffsetFetch [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21464:
URL: https://github.com/apache/kafka/pull/21464#discussion_r2823892310
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -552,24 +557,99 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
log.warn("A duplicated, inflight, request was identified, but unable to find it in the " +
"outbound buffer: {}", fetchRequest);
}
- if (error == null) {
- maybeUpdateLastSeenEpochIfNewer(res);
- result.complete(res);
- } else {
- if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
- if (fetchRequest.isExpired()) {
- log.debug("OffsetFetch request for {} timed out and won't be retried anymore", fetchRequest.requestedPartitions);
- result.completeExceptionally(maybeWrapAsTimeoutException(error));
- } else {
- fetchRequest.resetFuture();
- fetchOffsetsWithRetries(fetchRequest, result);
- }
- } else
- result.completeExceptionally(error);
+
+ // Success: no group-level error and no partition-level errors
+ if (error == null && !res.hasRetriablePartitionErrors()) {
Review Comment:
agree, updated 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