KAFKA-8729, pt 3: Add broker-side logic to handle the case when there are record_errors and error_message by tuvtran · Pull Request #7167 · apache/kafka

tuvtran · 2019-08-06T17:36:42Z

All the changes are in ReplicaManager.appendToLocalLog and ReplicaManager.appendRecords. Also, replaced LogAppendInfo. unknownLogAppendInfoWithLogStartOffset with LogAppendInfo. unknownLogAppendInfoWithAdditionalInfo to include those 2 new fields

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

tuvtran · 2019-08-06T17:41:40Z

Depending on #7150 to be merged

…protocol

guozhangwang · 2019-09-23T19:37:56Z

            brokerTopicStats.allTopicsStats.failedProduceRequestRate.mark()
            error(s"Error processing append operation on partition $topicPartition", t)
-            (topicPartition, LogAppendResult(LogAppendInfo.unknownLogAppendInfoWithLogStartOffset(logStartOffset), Some(t)))
+            (topicPartition, LogAppendResult(LogAppendInfo.unknownLogAppendInfoWithAdditionalInfo(logStartOffset, List(), ""), Some(t)))


I think we can improve this by doing the following:

inside LogValidator, when throwing InvalidTimestampException and InvalidRecordException, encode the relative offset inside the exception object.

then here we can add that relative offset as a singleton to the response.

By doing this on the client side, we can improve our error messaging such that we only tries to drop that single record but not the whole batch.

…broker-logic

tuvtran · 2019-10-03T17:17:03Z

Ready for another round of reviews @guozhangwang . The failed tests passed on my machine so likely they're flaky

guozhangwang · 2019-10-04T18:18:32Z

retest this please.

guozhangwang

ping @hachikuji for another look.

hachikuji

Thanks, left a few comments.

tuvtran · 2019-10-07T22:43:50Z

I have added a new RecordValidationException that includes a list of ErrorRecord and InvalidTimestampException or InvalidRecordException object. Ready for another round of reviews @guozhangwang @hachikuji

hachikuji

Thanks, left a few more comments.

guozhangwang · 2019-10-08T22:09:01Z

retest this please

…broker-logic

…ge is not null anymore

tuvtran · 2019-10-09T15:53:06Z

All tests have run and succeeded. Please let me know what you think about the latest changes @guozhangwang @hachikuji

guozhangwang · 2019-10-09T20:32:45Z

+
+        val expectedOffset = expectedInnerOffset.getAndIncrement()
+
+        // inner records offset should always be continuous


Nice catch here.

guozhangwang · 2019-10-09T22:45:32Z

Just minor comment, otherwise LGTM! Thanks @tuvtran

… are record_errors and error_message (#7167) All the changes are in ReplicaManager.appendToLocalLog and ReplicaManager.appendRecords. Also, replaced LogAppendInfo.unknownLogAppendInfoWithLogStartOffset with LogAppendInfo.unknownLogAppendInfoWithAdditionalInfo to include those 2 new fields. Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

guozhangwang · 2019-10-10T21:45:31Z

Merged to trunk, also pushed to 2.4

tuvtran · 2019-10-10T21:50:05Z

Thanks @hachikuji @guozhangwang for the reviews!

…batch does have continuous incremental offsets #7167 added a check for non-incremental offsets in `assignOffsetsNonCompressed`, which is not applicable for message format V0 and V1. Therefore, I added a condition to disable the check if the record version precedes V2. Author: Tu Tran <tu@confluent.io> Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com> Closes #7628 from tuvtran/KAFKA-9080

…batch does have continuous incremental offsets apache#7167 added a check for non-incremental offsets in `assignOffsetsNonCompressed`, which is not applicable for message format V0 and V1. Therefore, I added a condition to disable the check if the record version precedes V2. Author: Tu Tran <tu@confluent.io> Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com> Closes apache#7628 from tuvtran/KAFKA-9080

…ds (#7612) Background: Currently, whenever a batch is dropped because ofInvalidRecordException or InvalidTimestampException, only the culprit record appears in ProduceResponse.PartitionResponse.recordErrors. However, after users try to resend that batch excluding the rejected message, the latter records are not guaranteed to be free of problems. Changes: To address this issue, I changed the function signature of validateKey, validateRecord and validateTimestamp to return a Scala's Option object. Specifically, this object will hold the reason/message the current record in iteration fails and leaves to the callers (convertAndAssignOffsetsNonCompressed, assignOffsetsNonCompressed, validateMessagesAndAssignOffsetsCompressed) to gathered all troubling records into one place. Then, all these records will be returned along with the PartitionResponse object. As a result, if a batch contains more than one record errors, users see exactly which records cause the failure. PartitionResponse.recordErrors is a list of RecordError objects introduced by #7167 which include batchIndex denoting the relative position of a record in a batch and message indicating the reason of failure. Gotchas: Things are particularly tricky when a batch has records rejected because of both InvalidRecordException and InvalidTimestampException. In this case, the InvalidTimestampException takes higher precedence. Therefore, the Error field in PartitionResponse will be encoded with INVALID_TIMESTAMP. Reviewers: Guozhang Wang <wangguoz@gmail.com>

Tu Tran added 2 commits August 1, 2019 11:54

Add error_records and error_message to PartitionResponse

9bcf5ed

Change CorruptCorrupException in LogTest to InvalidRecordException

abf342c

Tu Tran added 2 commits August 28, 2019 17:01

Merge branch 'trunk' of https://github.com/apache/kafka into kip-467-…

a42ef2f

…protocol

Merge branch 'trunk' of https://github.com/apache/kafka into kip-467

e0725e4

tuvtran force-pushed the kip-467-broker-logic branch from 37f1d53 to a42ef2f Compare September 4, 2019 20:34

add changes on server side

0d1b51e

guozhangwang reviewed Sep 23, 2019

View reviewed changes

Tu Tran added 7 commits October 1, 2019 13:49

Merge branch 'trunk' of https://github.com/apache/kafka into kip-467-…

0804cbb

…broker-logic

reverse some changes

c13d8d5

add errorRecords to InvalidRecordException and InvalidTimestampException

37f88cd

add singletonMap to validateRecord and validateTimestamp

8c01ca3

add two unit tests

ebb524a

add test for producing with invalid timestamp

a20df44

set -> setIfExists

cf6d29d

guozhangwang reviewed Oct 4, 2019

View reviewed changes

hachikuji reviewed Oct 4, 2019

View reviewed changes

Tu Tran added 2 commits October 7, 2019 14:46

address GH comments

dc31c79

clean up exceptions

4fcfd7a

tuvtran force-pushed the kip-467-broker-logic branch from 1d3c187 to 4fcfd7a Compare October 7, 2019 21:59

fix relative vs absolute offset problem

2550341

hachikuji reviewed Oct 7, 2019

View reviewed changes

address review comments

0b9fd27

Tu Tran added 3 commits October 8, 2019 15:13

add comments and fix null pointer in PartitionResponse

ab9367a

Merge branch 'trunk' of https://github.com/apache/kafka into kip-467-…

d78be5a

…broker-logic

remove redundant fail statement and fix comment

877cde8

tuvtran force-pushed the kip-467-broker-logic branch from 28d75a0 to 877cde8 Compare October 8, 2019 22:30

Change testProduceWithInvalidTimestamp because the global error messa…

ca821c9

…ge is not null anymore

tuvtran changed the title ~~KAFKA-8729, pt 3: Add broker-side logic to handle the case when there are error_records and error_message~~ KAFKA-8729, pt 3: Add broker-side logic to handle the case when there are record_errors and error_message Oct 9, 2019

guozhangwang reviewed Oct 9, 2019

View reviewed changes

guozhangwang merged commit f41a5c2 into apache:trunk Oct 10, 2019

This was referenced Oct 29, 2019

KAFKA-8729: Change PartitionResponse to include all troubling records #7612

Merged

KAFKA-9080: Addresses MessageFormatChangeTest.testCompatibilty with version 0.9.0.1 #7628

Closed


		val expectedOffset = expectedInnerOffset.getAndIncrement()

		// inner records offset should always be continuous

Conversation

tuvtran commented Aug 6, 2019

Committer Checklist (excluded from commit message)

Uh oh!

tuvtran commented Aug 6, 2019

Uh oh!

guozhangwang Sep 23, 2019

Choose a reason for hiding this comment

Uh oh!

tuvtran commented Oct 3, 2019

Uh oh!

guozhangwang commented Oct 4, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tuvtran commented Oct 7, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

guozhangwang commented Oct 8, 2019

Uh oh!

tuvtran commented Oct 9, 2019

Uh oh!

guozhangwang Oct 9, 2019

Choose a reason for hiding this comment

Uh oh!

Uh oh!

guozhangwang commented Oct 9, 2019

Uh oh!

guozhangwang commented Oct 10, 2019

Uh oh!

tuvtran commented Oct 10, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

tuvtran commented Oct 7, 2019 •

edited

Loading