Skip to content

KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable #8486

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 8 commits into from
Jun 5, 2020
Merged

KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable #8486

Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
fddb49e
avoid offset validation when the leader epoch is not reliable, and ch…
Apr 10, 2020
dbaa133
address comments
Jun 3, 2020
94e0911
address last comment
Jun 3, 2020
03407c4
fix further comments
Jun 3, 2020
a6c0785
address logging comments
Jun 3, 2020
6a66b1b
propagate exception in onSuccess callback
Jun 3, 2020
1a97ccc
fix KafkaConsumerTest
Jun 4, 2020
87ca234
remove unnecessary complete validation logic
Jun 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -822,19 +822,25 @@ public void onSuccess(OffsetForEpochResult offsetsResult) {
FetchPosition requestPosition = fetchPositions.get(respTopicPartition);

if (respEndOffset.hasUndefinedEpochOrOffset()) {
handleOffsetOutOfRange(requestPosition, respTopicPartition,
"Failed leader offset epoch validation for " + respEndOffset
+ " since no end offset larger than current fetch epoch was reported");
try {
handleOffsetOutOfRange(requestPosition, respTopicPartition,
"Failed leader offset epoch validation for " + requestPosition
+ " since no end offset larger than current fetch epoch was reported");
} catch (OffsetOutOfRangeException e) {
// Swallow the OffsetOutOfRangeException to finish all partitions validation.

@hachikuji hachikuji Jun 3, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't feel great about having this in the code, even if it's supposed to be temporary. I think we should just fix the exception propagation bug in this patch. It seems like it would be straightforward to do something similar to what is done in the onFailure path.

                    if (!(e instanceof RetriableException) && !cachedOffsetForLeaderException.compareAndSet(null, e)) {
                        log.error("Discarding error in OffsetsForLeaderEpoch because another error is pending", e);
                    }

Above may not be ideal, but at least it provides a way to propagate individual errors.

}
} else {
Optional<OffsetAndMetadata> divergentOffsetOpt = subscriptions.maybeCompleteValidation(
respTopicPartition, requestPosition, respEndOffset);
divergentOffsetOpt.ifPresent(
divergentOffset -> truncationWithoutResetPolicy.put(respTopicPartition, divergentOffset));
divergentOffset -> {
log.info("Detected log truncation with diverging offset: {}", divergentOffset);
Comment thread
abbccdda marked this conversation as resolved.
Outdated
truncationWithoutResetPolicy.put(respTopicPartition, divergentOffset);
});
}
});

if (!truncationWithoutResetPolicy.isEmpty()) {
Comment thread
abbccdda marked this conversation as resolved.
log.error("Detected log truncation with diverging offsets " + truncationWithoutResetPolicy);
throw new LogTruncationException(truncationWithoutResetPolicy);
}
}
Expand Down Expand Up @@ -1328,7 +1334,7 @@ private void handleOffsetOutOfRange(FetchPosition fetchPosition,
", for fetch offset: %d, " +
"root cause: %s",
offsetOutOfRangePartitions, fetchPosition.offset, reason);
log.error(errorMessage);
log.info(errorMessage);
throw new OffsetOutOfRangeException(errorMessage, offsetOutOfRangePartitions);
}
}
Expand Down