KAFKA-5097: guard against unassigned partitions by enothereska · Pull Request #2876 · apache/kafka

enothereska · 2017-04-20T09:26:37Z

No description provided.

enothereska · 2017-04-20T09:28:30Z

@ijuma @hachikuji in trunk this has been fixed, but not in 0.10.2. Results in occasional stream test failure. A separate PR updates the stream tests and triggers the failure all the time #2867 (but can be ignored for now since problem manifests itself even without updated tests, although only occassionally). Note that Fetcher.java has diverged quite a bit in trunk from 0.10.2, so a full cherrypicking is hard. This just fixes the immediate problem.

ijuma · 2017-04-20T09:34:48Z

                    TopicPartition partition = nextInLineRecords.partition;
                    fetchedPartition = partition;
-                    fetchedOffsets = subscriptions.position(partition);
+                    if (subscriptions.isAssigned(partition)) {


I think we need to skip all the code in this else block if the partition is no longer assigned.

I'm not 100% sure. In the trunk code https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L523 we seem to drain the records (but the logic is slightly different, in that we drain them, but don't process them). Any suggestions welcome since I'm not super familiar with this code.

I took a closer look and it seems like this is a regression from a267316#diff-b45245913eaae46aa847d2615d62cde0R490 so we may need another RC.

Maybe we need to move the position call to inside !records.isEmpty since drainRecords does the check. But not sure, better to wait for @hachikuji to take a look.

cc @becketqin @lindong28 @hachikuji

Maybe we can go with this solution since it seems pretty safe. Shall we just remove the log.debug since we'll get that from the drainRecords call anyway? And let's run the system tests on this branch after that.

@enothereska Looking at the previous patch, it seems like the scope of the try/catch is unnecessarily large. I think it only needs to cover the call to parseCompletedFetch. If we do that, then there should be no need to access subscriptions.position down this path.

Mentioned offline, but a better solution (if Becket is right that the try/catch needs to cover this branch) would be the following:

fetchedOffsets = nextInLineRecords.fetchOffset;

Also, seems this variable shouldn't be plural.

@enothereska The trunk code does not need to access subscription.position, instead it uses PartitionRecords.nextInlineOffsets which should be the same as position because the position is updated to this value every time after a successful fetchRecords().

The big try/catch is to make sure the the exception from fetchRecords will also be caught and not result in loss of non-empty fetched.

so @becketqin to confirm, you are saying @hachikuji's suggestion on fetchedOffsets = nextInLineRecords.fetchOffset; is the way to go?
I ask because it takes 7 hours to run system tests :)

@enothereska Yes, I think that is a better solution. But I think @hachikuji was right that we don't need to cover both if/else branches. We just need to cover the parseCompletedFetch

Ok, stay tuned, making that change with next commit.

asfbot · 2017-04-20T10:13:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3051/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-04-20T10:13:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3046/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-04-20T10:16:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3047/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-20T11:39:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3050/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-20T11:40:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3054/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-04-20T11:41:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3049/
Test FAILed (JDK 7 and Scala 2.10).

enothereska · 2017-04-20T11:53:40Z

Streams system tests passed: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/262/ & https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/263/

enothereska · 2017-04-20T13:07:37Z

I believe unit test failure is known problem unrelated to PR, checking with @mjsax. Indeed it is fixed in trunk and is not related to PR. 92b7d75

enothereska · 2017-04-20T13:16:25Z

Retest this please

ijuma

LGTM, thanks. Will wait for @hachikuji's review.

asfbot · 2017-04-20T14:06:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3054/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-20T14:35:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3055/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-20T14:36:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3059/
Test PASSed (JDK 8 and Scala 2.11).

becketqin · 2017-04-20T17:50:54Z

Sorry for the regression. It was my bad. Will submit a PR next when backporting a patch does not merge cleanly.

@hachikuji The reason to have the big try catch is that some exceptions such as InvalidRecordException can be thrown from drainRecords() as well. So the previous patch was trying to cover that as well.

The fix LGTM.

hachikuji · 2017-04-20T17:55:53Z

@becketqin Are you sure about that? I must be missing it, can you point me to the line?

becketqin · 2017-04-20T18:25:07Z

@hachikuji You are right. This part of the code is different from trunk. We already parsed all the records in parseCompletedFetch. It was different from the trunk code.

enothereska · 2017-04-20T18:47:26Z

@hachikuji @becketqin @ijuma done. Restarting system tests. Btw, with previous change, all tests passed.

asfbot · 2017-04-20T19:37:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3064/
Test PASSed (JDK 7 and Scala 2.10).

hachikuji

LGTM. Good catch and thanks for the fix!

enothereska · 2017-04-20T19:47:39Z

Streams system tests pass: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/265/ & https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/266/

enothereska · 2017-04-20T19:48:40Z

Full test running here: https://jenkins.confluent.io/job/system-test-kafka-branch-builder-2/267/

…entially unassigned partitions Author: Eno Thereska <eno.thereska@gmail.com> Reviewers: Ismael Juma <ismael@juma.me.uk>, Jiangjie Qin <becket.qin@gmail.com>, Jason Gustafson <jason@confluent.io> Closes #2876 from enothereska/KAFKA-4755-fetcher-only

hachikuji · 2017-04-20T21:01:14Z

@enothereska I merged this to 0.10.2. Would you mind closing the PR since it looks like the merge doesn't take care of this automatically.

asfbot · 2017-04-20T21:47:29Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3069/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-04-20T21:47:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3065/
Test FAILed (JDK 8 and Scala 2.12).

Guard against unassigned partitions

93dfd5a

ijuma reviewed Apr 20, 2017

View reviewed changes

Remove debug msg

6bde1de

ijuma approved these changes Apr 20, 2017

View reviewed changes

enothereska changed the title ~~HOTFIX: guard against unassigned partitions~~ KAFKA-5097: guard against unassigned partitions Apr 20, 2017

Jason's suggestion

e2c83a5

hachikuji approved these changes Apr 20, 2017

View reviewed changes

enothereska closed this Apr 20, 2017

enothereska deleted the KAFKA-4755-fetcher-only branch April 20, 2017 21:39

Conversation

enothereska commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

ijuma Apr 20, 2017 • 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!

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!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

enothereska commented Apr 20, 2017

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

becketqin commented Apr 20, 2017

Uh oh!

hachikuji commented Apr 20, 2017

Uh oh!

becketqin commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

enothereska commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017

Uh oh!

hachikuji commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

asfbot commented Apr 20, 2017

Uh oh!

enothereska commented Apr 20, 2017 •

edited

Loading

ijuma Apr 20, 2017 •

edited

Loading

enothereska commented Apr 20, 2017 •

edited

Loading

enothereska commented Apr 20, 2017 •

edited

Loading