KAFKA-5682: Include partitions in exceptions raised during consumer record deserialization/validation #5410
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ionException in consumers Consumers would previously raise `SerializationException` with a message recommending the user seek past the failed offset at the given partition. The partition and offset was not given to the user and he had no way to know which partition caused the errors apart from manually parsing the message. This new exception type contains the topic, partition and offset data in it
…ptRecordException` (public) Also replace some usages to raise `CorruptRecordException`. this will allow subsequently raised `KafkaException` to add the exception as the cause
|
@hachikuji Please review when possible |
| else if (cachedRecordException instanceof RecordDeserializationException) | ||
| corruptOffset = ((RecordDeserializationException) cachedRecordException).offset(); | ||
| else | ||
| corruptOffset = nextFetchOffset; |
There was a problem hiding this comment.
Not absolutely sure if setting it to the nextFetchOffset is correct, but I figured it's the only thing to do.
We enter this if (corruptLastRecord) block if Fetcher#nextFetchedRecord() throws
…tion` when consumer encounters corrupted/invalid record This new exception contains the partition and offset information, allowing the user to skip past the record
* Make `RecordDeserializationException` and `InoperativeRecordException` internal classes * Add new interface that exposes topic and partition for the given exception
This reverts commit 3567a3c.
…erializationException` for `FaultyRecordException`
|
Closing in favor of #7499 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
JIRA: KAFKA-5682
KIP: KIP-334
Mailing Thread: thread
RecordDeserializationException- it extendsSerializationExceptionfor backwards compatibility. Throw that exception where appropriate with attached partition/offset.InoperativeRecordException- it extendsKafkaExceptionfor backwards compatibility. Throw that exception where appropriate with attached partition/offset.UnconsumableRecordException.InoperativeRecordExceptionandRecordDeserializationExceptionimplement itInvalidRecordExceptionextendApiException. It previously extendedCorruptRecordExceptionand thrown only for corrupt record scenarios. (as it's not in a public package).KafkaConsumer#poll()Committer Checklist (excluded from commit message)