KAFKA-7672: Restoring tasks need to be closed upon task suspension by guozhangwang · Pull Request #6113 · apache/kafka

guozhangwang · 2019-01-10T01:23:12Z

In activeTasks.suspend, we should also close all restoring tasks as well. Closing restoring tasks would not require task.close as in closeNonRunningTasks , since the topology is not initialized yet, instead only state stores are initialized. So we only need to call task.closeStateManager.
Also add @linyli001 's fix.
Unit tests updated accordingly.

Committer Checklist (excluded from commit message)

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

…sk-resumption

guozhangwang · 2019-01-10T23:55:13Z

        needsRestoring.clear();
        endOffsets.clear();
        needsInitializing.clear();
+        completedRestorers.clear();


This is @linyli001 's fix.

added some hack code to listeners and transformer init func before next version release..

guozhangwang · 2019-01-10T23:57:18Z

Triggered system test: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2255/

guozhangwang · 2019-01-10T23:59:41Z

        return restoringByPartition.get(partition);
    }

+    @Override


The following three functions do not have logical changes, just re-grouping all overridden functions on top of AssignedTasks here.

guozhangwang · 2019-01-11T00:01:59Z

            throw new IllegalStateException(logPrefix + "consumer has not been initialized while adding stream tasks. This should not happen.");
        }

-        changelogReader.reset();


We should not reset changelogReader since we still need its restored offset when closing the restoring tasks.

…sk-resumption

mjsax · 2019-01-14T17:41:42Z

+        // close all restoring tasks as well and then reset changelog reader;
+        // for those restoring and still assigned tasks, they will be re-created
+        // in addStreamTasks.
+        firstException.compareAndSet(null, active.closeAllRestoringTasks());


Why do we close those tasks instead of just suspending them? Could we close them only if they are not re-assigned?

The life time of a task is:

created -> [initializeStateStores] -> restoring (writes to the initialized state stores) -> [initializeTopology] -> running -> [closeTopology] -> suspended -> [closeStateManager] -> dead

I.e. the restoring tasks do not have topology initialized at all, whereas suspend call is just trying to closeTopology.

Ok, but why can't we keep restoring task open than, and hope they get reassigned so we can continue restoring them?

Yes we can, the issue is that today we clear all the store-restorers hence there's an issue.

We can, of course do some optimizations like do not close restoring tasks, and also do not clear their corresponding restorers as well, but this is out of the scope of this PR and I want to address it separately.

cc @vvcephei

@guozhangwang Can you create a Jira to track this cleanup?

Yup: https://issues.apache.org/jira/browse/KAFKA-7985

guozhangwang · 2019-01-18T23:15:00Z

retest this please

vvcephei · 2019-02-12T20:44:21Z

+        final Iterator<StreamTask> restoringTaskIterator = restoring.values().iterator();
+        while (restoringTaskIterator.hasNext()) {
+            final StreamTask task = restoringTaskIterator.next();
+            log.debug("Closing restoring and not re-assigned task {}", task.id());


Sorry for my denseness... Why are these "not re-assigned"? They're part of a data structure called "assigned tasks", which seems to imply that they are assigned.

Terminology storm coming :)

The assigned tasks contains the following non-overlapping sets (see my other comment as well for details): created, restoring, running, suspended. Normally created should be empty since once a task is created it should move on transit to either restoring or running immediately. This function is called below for suspendTasksAndState. In which case we should:

created and suspended: no action, since it should be empty.

running: transit to suspended.

restoring: close immediately.

The comment itself is indeed misleading, I will change to Closing restoring task

vvcephei

LGTM, @guozhangwang

I did ask one question about a comment, and I share @mjsax 's question about why we need to fully close stores that may be reassigned.

…sk-resumption

guozhangwang · 2019-02-22T18:45:15Z

Failed tests are not relevant. Local run passed.

mjsax

LGTM.

guozhangwang · 2019-02-22T18:49:47Z

retest this please

@linyli001

…6113) * In activeTasks.suspend, we should also close all restoring tasks as well. Closing restoring tasks would not require `task.close` as in `closeNonRunningTasks `, since the topology is not initialized yet, instead only state stores are initialized. So we only need to call `task.closeStateManager`. * Also add @linyli001 's fix. * Unit tests updated accordingly. Reviewers: Matthias J. Sax <mjsax@apache.org>, John Roesler <john@confluent.io>

guozhangwang · 2019-02-22T18:51:30Z

Cherry-picked to 2.2 as well.

* AK/trunk: (36 commits) KAFKA-7962: Avoid NPE for StickyAssignor (apache#6308) Address flakiness of CustomQuotaCallbackTest#testCustomQuotaCallback (apache#6330) KAFKA-7918: Inline generic parameters Pt. II: RocksDB Bytes Store and Memory LRU Caches (apache#6327) MINOR: fix parameter naming (apache#6316) KAFKA-7956 In ShutdownableThread, immediately complete the shutdown if the thread has not been started (apache#6218) MINOR: Refactor replica log dir fetching for improved logging (apache#6313) [TRIVIAL] Remove unused StreamsGraphNode#repartitionRequired (apache#6227) MINOR: Increase produce timeout to 120 seconds (apache#6326) KAFKA-7918: Inline generic parameters Pt. I: in-memory key-value store (apache#6293) MINOR: Fix line break issue in upgrade notes (apache#6320) KAFKA-7972: Use automatic RPC generation in SaslHandshake MINOR: Enable capture of full stack trace in StreamTask#process (apache#6310) KAFKA-7938: Fix test flakiness in DeleteConsumerGroupsTest (apache#6312) KAFKA-7937: Fix Flaky Test ResetConsumerGroupOffsetTest.testResetOffsetsNotExistingGroup (apache#6311) MINOR: Update docs to say 2.2 (apache#6315) KAFKA-7672 : force write checkpoint during StreamTask #suspend (apache#6115) KAFKA-7961; Ignore assignment for un-subscribed partitions (apache#6304) KAFKA-7672: Restoring tasks need to be closed upon task suspension (apache#6113) KAFKA-7864; validate partitions are 0-based (apache#6246) KAFKA-7492 : Updated javadocs for aggregate and reduce methods returning null behavior. (apache#6285) ...

@linyli001

…pache#6113) * In activeTasks.suspend, we should also close all restoring tasks as well. Closing restoring tasks would not require `task.close` as in `closeNonRunningTasks `, since the topology is not initialized yet, instead only state stores are initialized. So we only need to call `task.closeStateManager`. * Also add @linyli001 's fix. * Unit tests updated accordingly. Reviewers: Matthias J. Sax <mjsax@apache.org>, John Roesler <john@confluent.io>

guozhangwang added 4 commits January 9, 2019 17:22

first commit

f8e418c

close restoring tasks on suspend

cd7ec98

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

3734744

…sk-resumption

unit tests

178bcc3

guozhangwang changed the title ~~[WIP] Restoring tasks need to be closed~~ KAFKA-7672: Restoring tasks need to be closed Jan 10, 2019

guozhangwang changed the title ~~KAFKA-7672: Restoring tasks need to be closed~~ KAFKA-7672: Restoring tasks need to be closed upon task suspension Jan 10, 2019

guozhangwang commented Jan 10, 2019

View reviewed changes

guozhangwang commented Jan 11, 2019

View reviewed changes

guozhangwang requested a review from mjsax January 11, 2019 00:02

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

6832e3f

…sk-resumption

mjsax added the streams label Jan 14, 2019

mjsax reviewed Jan 14, 2019

View reviewed changes

vvcephei mentioned this pull request Feb 12, 2019

KAFKA-7672 : force write checkpoint during StreamTask #suspend #6115

Merged

3 tasks

vvcephei reviewed Feb 12, 2019

View reviewed changes

guozhangwang added 2 commits February 21, 2019 16:42

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

4429819

…sk-resumption

github comments

912a456

mjsax approved these changes Feb 22, 2019

View reviewed changes

guozhangwang merged commit 0d461e4 into apache:trunk Feb 22, 2019

guozhangwang deleted the K7672-task-resumption branch April 24, 2020 23:59

Conversation

guozhangwang commented Jan 10, 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!

Choose a reason for hiding this comment

Uh oh!

suiyuan2009 Feb 7, 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!

guozhangwang commented Jan 10, 2019

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!

guozhangwang Feb 22, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 18, 2019 • 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!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Feb 22, 2019

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Feb 22, 2019

Uh oh!

guozhangwang commented Feb 22, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

guozhangwang commented Jan 10, 2019 •

edited

Loading

suiyuan2009 Feb 7, 2019 •

edited

Loading

guozhangwang Feb 22, 2019 •

edited

Loading

guozhangwang commented Jan 18, 2019 •

edited

Loading