HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed by ableegoldman · Pull Request #8140 · apache/kafka

ableegoldman · 2020-02-20T02:57:14Z

This fixes two issues which together caused the soak to crash/some test to fail occasionally.

What happened was:
In the main StreamThread loop we initialized a new task in TaskManager#checkForCompletedRestoration which includes registering, but not initializing, its changelogs. We then complete the loop and call poll, which resulted in a rebalance that revoked the newly-initialized task. In TaskManager#handleAssignment we then closed the task cleanly and go to remove the changelogs from the StoreChangelogReader only to get an IllegalStateException because the changelog partitions were not in the restore consumer's assignment (due to being uninitialized).

This by itself should^ be a recoverable error, as we catch exceptions here and retry closing the task as unclean. Of course the task actually was successfully closed (clean) so we now get an unexpected exception Illegal state CLOSED while closing active task

The fix(es) I'd propose are:

~~Keep the restore consumer's assignment in sync with the registered changelogs, ie the set ChangelogReader#changelogs but pause them until they are initialized~~ edit: since the consumer does still perform some actions (gg fetches) on paused partitions, we should avoid adding uninitialized changelogs to the restore consumer's assignment. Instead, we should just skip them when removing.
Move the StoreChangelogReader#remove call to before the task.closeClean so that the task is only marked as closed if everything was successful. We should do so regardless, as we should (attempt to) remove the changelogs even if the clean close failed and we must do unclean.

…changelogs for initialized tasks

ableegoldman · 2020-02-20T02:57:30Z

@guozhangwang @vvcephei

ableegoldman · 2020-02-20T02:58:03Z


                transitionTo(State.SUSPENDED);
-                log.info("Suspended running");
+                log.info("Suspended active");


Technically it may be running or restoring here

ableegoldman · 2020-02-20T03:00:17Z


+        // we can always let changelog reader try restoring in order to initialize the changelogs;
+        // if there's no active restoring or standby updating it would not try to fetch any data
+        changelogReader.restore();


This isn't a necessary part of the fix, but I feel it makes more sense to move this here: the restore consumer's can only make progress on tasks that have been initialized, which only happens in taskManager.tryToCompleteRestoration

I agree. Originally I put it in front intentionally since initializeIfNeeded is called at task creation not here, and we would only try to initialize the changelogs in restore and hence only check if the restoration can be completed.

Now that we've delayed the initializeIfNeeded in tryToComplete we should reorder the restore call as well.

ableegoldman · 2020-02-20T03:01:17Z

     * @throws StreamsException if the store's change log does not contain the partition
     */
-    boolean checkForCompletedRestoration() {
+    boolean tryToCompleteRestoration() {


Also not necessary, but I feel it better suggests/reminds that it is taking some action here and not just checking a condition

vvcephei

Thanks, @ableegoldman !

Looks good to me. There's one logging mistake (below) that needs to be corrected, but otherwise I think it's ready to merge.

vvcephei · 2020-02-20T15:23:25Z

Ok to test.

guozhangwang

Hi @ableegoldman Thanks for the hot-fix! I agree with you about fix 2) here, for fix 1) I have some alternative ideas since it is risky to add the partitions without setting their positions (we discussed in another PR @vvcephei and decide we may want to set the reset-policy to always none for restore consumer as well, which would cause NoOffsetsForPartition in the immediate poll call.

guozhangwang · 2020-02-20T18:41:57Z

                ", this should not happen: " + changelogs);
        }
+
+        addChangelogsToRestoreConsumer(Collections.singleton(partition));


The reason I did not do this is that in consumer, even if a partition is paused, we would still try to fetch is fetching position in the poll call if it was not available. And that's why I have to defer this after we've found the starting position and initialize them with seek calls.

So I'm thinking maybe we would just remove the illegal-state check from the remoreFromRestoreConsumer call and just log it as a debug?

Ah, I didn't realize the consumer would do anything with paused partitions. I'm not sure we want to remove the verification entirely, but we can check any "missing" partitions against the changelogs map instead.

guozhangwang · 2020-02-20T18:47:10Z


+        // we can always let changelog reader try restoring in order to initialize the changelogs;
+        // if there's no active restoring or standby updating it would not try to fetch any data
+        changelogReader.restore();


I agree. Originally I put it in front intentionally since initializeIfNeeded is called at task creation not here, and we would only try to initialize the changelogs in restore and hence only check if the restoration can be completed.

Now that we've delayed the initializeIfNeeded in tryToComplete we should reorder the restore call as well.

guozhangwang · 2020-02-20T18:47:43Z

                try {
-                    task.closeClean();
                    changelogReader.remove(task.changelogPartitions());
+                    task.closeClean();


Yup! I also found this bug and fixed in another PR, but great to see it confirmed in yours as well :)

Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

ableegoldman · 2020-02-20T21:16:47Z

Ran the tests locally twice: one unrelated failure in shouldWorkWithRebalance (that fix coming next) but otherwise all green. Prior to this I saw 1-2 failures per run due to the issue(s) fixed here

guozhangwang

LGTM! Merging to trunk now.

@test

* apache-github/trunk: (23 commits) KAFKA-9530; Fix flaky test `testDescribeGroupWithShortInitializationTimeout` (apache#8154) HOTFIX: fix NPE in Kafka Streams IQ (apache#8158) MINOR: set scala version automatically based on gradle.properties KAFKA-9577; SaslClientAuthenticator incorrectly negotiates SASL_HANDSHAKE version (apache#8142) KAFKA-9441: Add internal TransactionManager (apache#8105) MINOR: Document endpoints for connector topic tracking (KIP-558) MINOR: Standby task commit needed when offsets updated (apache#8146) KAFKA-9206; Throw KafkaException on CORRUPT_MESSAGE error in Fetch response (apache#8111) MINOR: Remove unwanted regexReplace on tests/kafkatest/__init__.py KAFKA-9586: Fix errored json filename in ops documentation KAFKA-9575: Mention ZooKeeper 3.5.7 upgrade KAFKA-9481: Graceful handling TaskMigrated and TaskCorrupted (apache#8058) HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed (apache#8140) MINOR: Fix javadoc at org.apache.kafka.clients.producer.KafkaProducer.InterceptorCallback#onCompletion (apache#7337) MINOR: Improve EOS example exception handling (apache#8052) MINOR: Fix a number of warnings in clients test (apache#8073) MINOR: Update shell scripts to support z/OS system (apache#7913) MINOR: Wording fix in Streams DSL docs (apache#5692) MINOR: Add missing @test annotation to MetadataTest#testMetadataMerge (apache#8141) KAFKA-9533: ValueTransform forwards `null` values (apache#8108) ...

@test

…etrics-common * confluent/master: (76 commits) KAFKA-9530; Fix flaky test `testDescribeGroupWithShortInitializationTimeout` (apache#8154) HOTFIX: fix NPE in Kafka Streams IQ (apache#8158) MINOR: set scala version automatically based on gradle.properties KAFKA-9577; SaslClientAuthenticator incorrectly negotiates SASL_HANDSHAKE version (apache#8142) KAFKA-9441: Add internal TransactionManager (apache#8105) MINOR: Document endpoints for connector topic tracking (KIP-558) MINOR: Standby task commit needed when offsets updated (apache#8146) Changes to migrate to Artifactory (#263) KAFKA-9206; Throw KafkaException on CORRUPT_MESSAGE error in Fetch response (apache#8111) MINOR: Remove unwanted regexReplace on tests/kafkatest/__init__.py KAFKA-9586: Fix errored json filename in ops documentation KAFKA-9575: Mention ZooKeeper 3.5.7 upgrade KAFKA-9481: Graceful handling TaskMigrated and TaskCorrupted (apache#8058) HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed (apache#8140) MINOR: Fix javadoc at org.apache.kafka.clients.producer.KafkaProducer.InterceptorCallback#onCompletion (apache#7337) MINOR: Improve EOS example exception handling (apache#8052) MINOR: Fix a number of warnings in clients test (apache#8073) MINOR: Update shell scripts to support z/OS system (apache#7913) MINOR: Wording fix in Streams DSL docs (apache#5692) MINOR: Add missing @test annotation to MetadataTest#testMetadataMerge (apache#8141) ...

apurvam · 2020-02-25T18:48:30Z

Why are there no new test cases for this bug in this PR?

ableegoldman · 2020-02-25T18:55:08Z

I'll make a note to add test cases for this in the next PR

ableegoldman added 2 commits February 19, 2020 15:15

try removing partitions first, log exception

24327a5

keep restore consumer's assignment in sync with StoreChangelogReader#…

e110d7c

…changelogs for initialized tasks

ableegoldman commented Feb 20, 2020

View reviewed changes

vvcephei reviewed Feb 20, 2020

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java Outdated

guozhangwang reviewed Feb 20, 2020

View reviewed changes

ableegoldman and others added 2 commits February 20, 2020 12:13

Github suggestion: logging

566dcc1

Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

dont assign uninitialized changelogs, dont try to remove them

3164fce

ableegoldman changed the title ~~HOTFIX: keep restore consumer assignment in sync & don't prematurely mark task closed~~ HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed Feb 20, 2020

guozhangwang approved these changes Feb 20, 2020

View reviewed changes

guozhangwang merged commit 0d16c26 into apache:trunk Feb 20, 2020

ableegoldman deleted the HOTFIX-dont-re-close-closed-task branch June 26, 2020 22:37

Conversation

ableegoldman commented Feb 20, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ableegoldman commented Feb 20, 2020

Uh oh!

ableegoldman Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

ableegoldman Feb 20, 2020 • 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!

guozhangwang Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

ableegoldman Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

vvcephei commented Feb 20, 2020

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

ableegoldman Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

guozhangwang Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

guozhangwang Feb 20, 2020

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Feb 20, 2020

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

apurvam commented Feb 25, 2020

Uh oh!

ableegoldman commented Feb 25, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

ableegoldman commented Feb 20, 2020 •

edited

Loading

ableegoldman Feb 20, 2020 •

edited

Loading