KAFKA-8729, pt 2: Add error_records and error_message to PartitionResponse by tuvtran · Pull Request #7150 · apache/kafka

tuvtran · 2019-08-01T18:57:07Z

As noted in the KIP-467, the updated ProduceResponse is

Produce Response (Version: 8) => [responses] throttle_time_ms
  responses => topic [partition_responses]
    topic => STRING
    partition_responses => partition error_code base_offset log_append_time log_start_offset
      partition => INT32
      error_code => INT16
      base_offset => INT64
      log_append_time => INT64
      log_start_offset => INT64
      error_records => [INT32]         // new field, encodes the relative offset of the records that caused error
      error_message => STRING          // new field, encodes the error message that client can use to log itself
    throttle_time_ms => INT32

with a new error code:

INVALID_RECORD(86, "Some record has failed the validation on broker and hence be rejected.", InvalidRecordException::new);

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:14Z

@guozhangwang could you take a look at this?

…protocol

tuvtran · 2019-09-03T17:20:36Z

retest this please

guozhangwang

Made a pass over this PR. Also ping @rayokota @hachikuji for another look.

guozhangwang · 2019-09-05T19:15:55Z


  override def toString = s"[acksPending: $acksPending, error: ${responseStatus.error.code}, " +
-    s"startOffset: ${responseStatus.baseOffset}, requiredOffset: $requiredOffset]"
+    s"startOffset: ${responseStatus.baseOffset}, requiredOffset: $requiredOffset, " +


I think we do not need to print them here since if the error records / message are ever not null / none there will be no delayer produce generated.

guozhangwang · 2019-09-05T20:39:48Z

+                                    "If LogAppendTime is used for the topic, the timestamp will be the broker local " +
+                                    "time when the messages are appended."),
+                            LOG_START_OFFSET_FIELD,
+                            new Field(ERROR_RECORDS_KEY_NAME, new ArrayOf(INT32)),


Why we do not have a description string for error records as well?

guozhangwang · 2019-09-06T18:30:47Z

retest this please

hachikuji · 2019-09-06T21:27:01Z

+import org.apache.kafka.common.errors.ApiException;

-public class InvalidRecordException extends CorruptRecordException {
+public class InvalidRecordException extends ApiException {


Hmm.. Isn't this an incompatible change? Does this exception not get exposed anywhere currently? Maybe it would be better to deprecate common/record/InvalidRecordException, but still leave it around.

Yeah I think InvalidRecordException was not exposed as a public API before, and now we are moving its package as well to make it a public class.

…ordTest

hachikuji

Left a few comments. It's worth considering this in the context of KIP-482. I think we would make these fields optional?

guozhangwang · 2019-09-09T21:20:35Z

I'd recommend we do not block this PR on KIP-482, since that is not merged in yet. If KIP-482 did gets in before the release, we can have another PR incorporating the feature and declare those additional fields as optional.

guozhangwang · 2019-09-09T21:23:28Z

Other than @hachikuji 's comments and the failed jenkins test, this PR LGTM.

tuvtran · 2019-09-10T21:47:24Z

kafka.api.SaslOAuthBearerSslEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl is flaky (it passed on my machine).

…d DefaultRecordBatch

tuvtran · 2019-09-11T17:43:03Z

@guozhangwang I noticed that in DefaultRecordBatch and Fetcher we still use InvalidRecordException to check if a record (batch) is valid or not, so I changed those to use CorruptRecordException

guozhangwang · 2019-09-11T18:10:27Z

@guozhangwang I noticed that in DefaultRecordBatch and Fetcher we still use InvalidRecordException to check if a record (batch) is valid or not, so I changed those to use CorruptRecordException

SGTM, throwing and catching CorruptRecordException is a better.

Maybe do a quick call-trace checking to make sure such exception thrown changes does not affect any public facing functions which would then throw different exceptions as well -- I did a quick look and it seems callers / callees are all internal classes, but always better to have another pair of eyes.

…protocol

guozhangwang

Nit comments, otherwise LGTM. cc @hachikuji

guozhangwang · 2019-09-19T01:47:52Z

     * {@link Errors#NOT_LEADER_FOR_PARTITION} (6)
     * {@link Errors#MESSAGE_TOO_LARGE} (10)
-     * {@link Errors#INVALID_TOPIC} (17)
+     * {@link Errors#INVALID_TOPIC_EXCEPTION} (17)


Why we want to add the _EXCEPTION suffix? Others do not have it.

Someone actually changed it from INVALID_TOPIC to INVALID_TOPIC_EXCEPTION: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java#L162

Got it, I think it was in this case long time ago, we can fix it in another PR if we want.

guozhangwang · 2019-09-19T01:49:45Z

+        ]},
        { "name":  "ErrorMessage", "type":  "string", "versions": "8+", "default": "", "ignorable":  true,
-          "about":  "The error message of the records that cause the batch to be dropped"}
+          "about":  "The error message of records that caused the batch to be dropped"}


nit: How about The global error message summarizing the common root cause of the records that makes the batch to be dropped.

hachikuji

Thanks, left a few comments.

hachikuji · 2019-09-20T20:38:31Z

+          "about": "The relative offsets of records that caused the batch to be dropped", "fields": [
+          { "name":  "RelativeOffset", "type": "int32", "versions":  "8+",
+            "about":  "The relative offset of the record that cause the batch to be dropped" },
+          { "name": "RelativeOffsetErrorMessage", "type": "string", "versions":  "8+",


Do we want this to be nullable? Same for ErrorMessage below. We'll need to make a similar change to the old style spec.

The error message is great, but it can't be used programmatically on the client/app, so we'll also need the 16-bit error code here.

@edenhill I've thought about this when proposing KIP-467 and the reason I did not try to encode the 16 bit error code is that:

At the moment, from all the possible errors that can be returned from the broker, except the InvalidRecordException all others would cause the response to be returned immediately, which means that we would not have more than one error code that would trigger the response to be rejected.

For the only possible InvalidRecordException that may be co-exist with other exceptions in the rejected response, that would cause the error records to be trimmed, whereas all other error codes would still cause the whole batch to be dropped. Therefore we've decided to make the broker-side logic to execute the following: when any non-InvalidRecordException is thrown, since that would cause the whole batch to be dropped we would just return that error code on the per-partition and ignore any other InvalidRecordException; if the only exception thrown is InvalidRecordException on specific record(s), then we encode it on per-partition level.

That being said, if people feel that in the future or even now we would like to change the broker-side logic and it's worth paying the 16bits in addition to 8bit for encoding an empty string, then I'm fine doing that in this PR as well.

@hachikuji @tuvtran @junrao please let me know.

Thanks for explaining your reasoning @guozhangwang.

I think the key here is "At the moment": having an error code field in this response would allow any future error code to be added without having to bump the protocol version, more specifically the range of returned error codes for the per-message errors should be left undefined as this is a means for communication between some future message verification mechanism on the broker and the application producing the data, rather than between the broker and the producer.

For space efficiency I'd suggest using a varint, but I doubt it matters in practice since these errors should be rare, and the two bytes of error code are just a fraction of the error message. And the erroneous Produce response is most likely far smaller than the request, or a fetch response for that matter.

@edenhill Honestly I'm still not convinced of the future usage of per-record error codes -- atm we have not observed concrete error codes that would split within a single batch that would like to be handled differently by clients. If we do find out that become a common scenario to support we can still always bump up the version, which is not a very costly evolution anyways. On the other hand, since we do not have varint in place yet we would still need to encode 2bytes with the empty string: note that the per-record error message would be commonly empty in practice also.

Another thing worth mentioning is that even if we add the fields now, any new error code beyond InvalidRecord that we'd like to add in the future would still require the clients to upgrade because otherwise they cannot recognize them anyways.

@guozhangwang Okay, sounds good, I'll rest my case.

hachikuji · 2019-09-20T20:51:58Z

        }

        public PartitionResponse(Errors error, long baseOffset, long logAppendTime, long logStartOffset) {
+            this(error, baseOffset, logAppendTime, logStartOffset, new HashMap<>(), "");


nit: use Collections.emptyMap()?

…omments

tuvtran · 2019-09-24T06:17:59Z

retest this please

tuvtran · 2019-09-25T21:47:09Z

retest this please

…protocol

tuvtran · 2019-09-30T17:59:39Z

retest this please

guozhangwang

Made another pass, LGTM!

guozhangwang · 2019-10-01T02:27:38Z

                if (partStruct.hasField(LOG_APPEND_TIME_KEY_NAME))
                    partStruct.set(LOG_APPEND_TIME_KEY_NAME, part.logAppendTime);
                partStruct.setIfExists(LOG_START_OFFSET_FIELD, part.logStartOffset);
+


nit: we can probably do some cleanup in pr.3, e.g. line 269 above can be replaced with setIfExist as well.

cc @tuvtran

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

This was referenced Aug 6, 2019

KAFKA-8729, pt 3: Add broker-side logic to handle the case when there are record_errors and error_message #7167

Merged

[WIP] KAFKA-8729, pt 4: Add client-side logic to augment error handling and partially fail records specified in error_records #7168

Closed

tuvtran changed the title ~~KAFKA 8729, pt 2: Add error_records and error_message to PartitionResponse~~ KAFKA-8729, pt 2: Add error_records and error_message to PartitionResponse Aug 28, 2019

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

guozhangwang reviewed Sep 5, 2019

View reviewed changes

address GH comments

460a3b4

hachikuji reviewed Sep 6, 2019

View reviewed changes

change InvalidRecordException to CorruptRecordException in LegacyRec…

61285e4

…ordTest

hachikuji reviewed Sep 9, 2019

View reviewed changes

address Jason's comments

6288ad0

Tu Tran added 2 commits September 10, 2019 14:49

fix minor issue in ProduceRequestTest

501a824

Change InvalidRecordException to CorruptRecordException in Fetcher an…

4f03549

…d DefaultRecordBatch

Tu Tran added 3 commits September 16, 2019 15:06

bump the error code to 87

ccb559f

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

3b8ece1

…protocol

modify error_records from [int32] to map[int32, string]

3535b69

guozhangwang reviewed Sep 19, 2019

View reviewed changes

change global error message

4b830e5

hachikuji reviewed Sep 20, 2019

View reviewed changes

add tests in messageTest and RequestResponseTest and addressed GH's c…

d34f1b7

…omments

Tu Tran added 2 commits September 24, 2019 09:47

fix checkstyle

74ae763

null pointer

18deeb4

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

1de5dc0

…protocol

guozhangwang approved these changes Oct 1, 2019

View reviewed changes

guozhangwang merged commit f6f24c4 into apache:trunk Oct 1, 2019

chia7712 mentioned this pull request Jul 9, 2024

KAFKA-16689: Move LogValidatorTest to storage module #16167

Merged

3 tasks

Conversation

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

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

tuvtran commented Aug 6, 2019

Uh oh!

tuvtran commented Sep 3, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

guozhangwang commented Sep 6, 2019

Uh oh!

hachikuji Sep 6, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

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!

guozhangwang commented Sep 9, 2019

Uh oh!

guozhangwang commented Sep 9, 2019

Uh oh!

tuvtran commented Sep 10, 2019

Uh oh!

tuvtran commented Sep 11, 2019

Uh oh!

guozhangwang commented Sep 11, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

tuvtran commented Aug 1, 2019 •

edited

Loading

hachikuji Sep 6, 2019 •

edited

Loading