KAFKA-9113: P4, Refactor Task Lifecycle by vvcephei · Pull Request #6 · guozhangwang/kafka

vvcephei · 2020-01-15T22:59:16Z

@guozhangwang : I'm in the middle of this implementation, but here's what I have so far, if you're curious.

Committer Checklist (excluded from commit message)

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

…ager

guozhangwang

Thanks @vvcephei ! I continued work on standby task based on your branch, and also briefly went through your PR and left some high-level comments. Most of the changes around StandbyTask are here: https://github.com/guozhangwang/kafka/pull/7/files#diff-7012e86d489a291404d83f961ad40217

guozhangwang · 2020-01-16T23:12:27Z

+        CREATED, RESTORING, RUNNING, REVOKED, CLOSED;
+
+
+        static void validateTransition(final State oldState, final State newState) {


Why do we want running -> running, restoring -> restoring etc? Is there a specific scenario that we'd like to cover?

I've removed them.

guozhangwang · 2020-01-20T01:11:02Z

-     * </pre>
-     */
+    @Override
+    public void suspend() {


Since StandbyTask do not have suspension / resumption since it should only have created / restoring / done states, I'd suggest we just remove it from Task interface and only have it in StreamTask.

I see. I was thinking that the only place we would suspend is in listener#onPartitionsRevoked where we only suspend active tasks, and similarly in assignor#onAssignment we only resume active tasks in "suspended" state, but I can buy the argument of easy testing, will make a thorough pass on TaskManager.

guozhangwang · 2020-01-20T01:24:08Z

     * @throws StreamsException if the store's change log does not contain the partition
     */
-    boolean initializeStateStores();
+    void initializeStateStores();


I think we can remove initializeStateStores / initializeTopology / initializeMetadata with transitionTo (the initializeIfNeeded can also be private):

for for both active and standby transitionTo(Created -> Restoring) always register stores; for active task we also initialize metadata.

for active TransitionTo(Restoring -> Running) we initialize the topology.

for standby there's no Transition to Running / Suspended, if it ever gets call we throw IllegalState.

…nto refactor-task-lifecycle

guozhangwang

@vvcephei I pushed a commit to cleanup the StandbyTaskTest, and left some more comments.

guozhangwang · 2020-01-21T22:50:30Z

+
+            // no topology needs initialized, we can transit to RUNNING
+            // right after registered the stores
+            transitionTo(State.RUNNING);


Right now we CREATED -> RUNNING is not valid as we expect it to always first transit to RESTORING. If we do not want to make task-specific transition rules (I'd prefer that) I suggest we make standby to be in RESTORING as its normal processing state.

We could, but it seems to muddy the waters. It seems easier to think of "restoring" as some up-front setup, while "running" is the steady state. Since standbys don't need any up-front setup, it seems more straightforward to skip over restoring than to skip over running.

guozhangwang · 2020-01-21T23:36:48Z

    @Override
-    public void close(final boolean clean) throws ProcessorStateException {
-        ProcessorStateException firstException = null;
+    public void close() throws ProcessorStateException {


Not a comment: the clean flag is not used inside state store anymore so I've removed it.

guozhangwang · 2020-01-21T23:45:06Z


    void commit();

+    void suspend();


How about removing suspend / resume from Task since they are only for StreamTask?

I added it as part of an effort to make the task management in general ignorant of the specific type of task. E.g., when the TaskManager calls this method, it's because it entered a rebalance and therefore wants to suspend all processing work. StandbyTasks know that they don't have any processing work, so they can implement this as a no-op, but the TaskManager doesn't need to worry about whether different kinds of tasks have processing work. It can just tell all tasks to suspend processing. Standby tasks can say (completely truthfully) "Ok, I've suspended all active processing (by doing nothing, since I never had any to begin with".

Okay, but in TaskManager I saw you still specifically loop over active tasks first then standby tasks :)

Aha! That was to invoke the active or standby task creator, I think... Of course we have to know what kind of task we need to construct so we can call the right constructor.

guozhangwang

Made a pass on the current PR (non-testing part only).

guozhangwang · 2020-01-22T01:27:47Z


    public static class InternalConfig {
        public static final String TASK_MANAGER_FOR_PARTITION_ASSIGNOR = "__task.manager.instance__";
+        public static final String STREAMS_METADATA_STATE_FOR_PARTITION_ASSIGNOR = "__streams.metadata.state.instance__";


Could you elaborate a bit on why we need to pass this to consumer as well?

We used to abuse the TaskManager to also be a "container" for the StreamsMetadataState object. But its purpose is unrelated, so I broke the dependency and just inject both classes into the StreamsPartitionAssignor.

Sounds good!

guozhangwang · 2020-01-22T01:28:57Z

        }
    }

+    void addSubscribedTopics(final List<TopicPartition> partitions, final String logPrefix) {


Where is this function used?

at the end of "handleAssignment" in the TaskManager. This whole function used to be in the TaskManager.

Got it, maybe rename to addSubscribedTopicsFromAssignment, and inline updateSubscribedTopics? Just a nit.

guozhangwang · 2020-01-22T02:05:51Z

-        final Exception exception = firstException.get();
-        if (exception != null) {
-            throw new StreamsException(logPrefix + "failed to suspend stream tasks", exception);
+        for (final TaskId taskId : revokedTasks) {


nit: maybe add a check that these tasks are actives only? Also do we guarantee that the task-manager would not host both active and standby of the same task-id?

I was uneasy about that early on, but it should be impossible for a single instance of KaflaStreams, let alone one of its threads, to host more than one copy of a single task, active or otherwise.

guozhangwang · 2020-01-22T02:06:25Z

-        return active.previousRunningTaskIds();
+        return tasks.values()
+                    .stream()
+                    .filter(t -> t.isActive() && t.state() == Task.State.SUSPENDED)


Is that right prev-running == current suspended? Just checking.

…/kafka into K9113-standby-task-p5

guozhangwang

Pushed a commit that fixed a bunch of bugs I found while debugging integration tests.

@vvcephei could you also add some unit test coverage so that we can capture those cases earlier?

guozhangwang · 2020-01-24T23:10:46Z

            log.error("Received error code {} - shutdown", streamThread.getAssignmentErrorCode());
            streamThread.shutdown();
        } else {
+            taskManager.handleRebalanceComplete();


The ordering within consumer coordinator is the following:

assignor#onAssignment

the assignment gets updated internally

listener#onPartitionsAssigned

So we have to call consumer.pause(all) it inside 3) since otherwise the assignment() is still stale.

guozhangwang · 2020-01-24T23:12:08Z

-                changelogOffsets.put(storeMetadata.changelogPartition, storeMetadata.offset);
+                // for changelog whose offset is unknown, use 0L indicating earliest offset
+                // otherwise return the current offset + 1 as the next offset to fetch
+                changelogOffsets.put(


note the offset inside state store is the last record's offset, but to return changelogOffsets used for comparing with other end offsets etc we need to return the next offset to process (and yes it is still vulnerable to txn marker cases but that's fine for lag info I think).

guozhangwang · 2020-01-24T23:13:24Z

-        } else {
-            throw new IllegalStateException("Expected only to resume from SUSPENDED state, but was " + state());
+        switch (state()) {
+            case RUNNING:


We may call assignor#onAssignment multiple times without onPartitionsRevoked, therefore not all existing active tasks are in SUSPENDED when we are handleAssignmenting. For those that are still in RUNNING / RESTORING we just do nothing.

guozhangwang · 2020-01-24T23:15:12Z

        // if the state is still in PARTITION_ASSIGNED after the poll call
        if (state == State.PARTITIONS_ASSIGNED) {
-            if (taskManager.initializeNewTasksAndCheckForCompletedRestoration()) {
+            if (taskManager.checkForCompletedRestoration()) {


Here I have to break the changelogReader#restore out of the TaskManager so that 1) we call it every iteration no matter if we are restoring active or updating standby; 2) we only need to initialize tasks once.

guozhangwang · 2020-01-24T23:16:03Z

-                    if (committed > 0) {
-                        final long commitLatency = advanceNowAndComputeLatency();
-                        commitSensor.record(commitLatency / (double) committed, now);
+        if (state == State.RUNNING) {


We should only process active tasks if we are in RUNNING state: note that there maybe some buffered records already even after a rebalance and so if we proceed we could mistakenly process them from the buffer even if we are not in RUNNING.

guozhangwang · 2020-01-24T23:16:33Z

            logPrefix
        );

+        // initialize the created tasks


See other comment above: we move the initialization logic right after creation so that within the iteration we only check for completion.

mjsax · 2020-01-30T22:04:07Z

 */
-public class StandbyTask extends AbstractTask {
+public class StandbyTask extends AbstractTask implements Task {
+    private final TaskId id;


Why do we move TaskId from AbstractTask to StandbyTask and also StreamTask -- seem we are duplicating code -- every task has an ID? Similar for other class members? Why not following best practice of the OO model?

Yeah I can move this to AbstractTask.

mjsax · 2020-01-30T22:07:23Z

-    private void updateThreadMetadata(final Map<TaskId, StreamTask> activeTasks,
-                                      final Map<TaskId, StandbyTask> standbyTasks) {
+    private void updateThreadMetadata(final Map<TaskId, Task> activeTasks,
+                                      final Map<TaskId, Task> standbyTasks) {


Why do we weaken the types here?

Because the task manager now maintains the active / standby tasks in a single collection so the exposed APIs only return a map of Tasks. Also for the caller there's no need to have the stronger types.

mjsax · 2020-01-30T22:08:29Z

+            }
+        };
+
+        public abstract boolean hasBeenRunning();


What is the semantic meaning of this method and why do we need it?

The main purpose of this was to avoid calling closeTopology duplicated times if the state has already been transited to CLOSING -- now that we've had this branching logic on the caller (StreamTask) I think we can remove this part.

mjsax · 2020-01-30T22:13:19Z

+        } else if (oldState == RESTORING && (newState == RUNNING || newState == SUSPENDED || newState == CLOSING)) {
+        } else if (oldState == RUNNING && (newState == SUSPENDED || newState == CLOSING)) {
+        } else if (oldState == SUSPENDED && (newState == RUNNING || newState == CLOSING)) {
+        } else if (oldState == CLOSING && (newState == CLOSING || newState == CLOSED)) {


This is a quite weird way to express a logical condition... Why not just use || ? Or even better, why not encode valid transitions in an enum type similar to existing code -> https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L137-L143

I've moved it to Task following the similar pattern of KafkaStreams, StreamThread. LMK WDYT.

Sounds good!

This PR is collaborated by Guozhang Wang and John Roesler. It is a significant tech debt cleanup on task management and state management, and is broken down by several sub-tasks listed below: Extract embedded clients (producer and consumer) into RecordCollector from StreamTask. guozhangwang#2 guozhangwang#5 Consolidate the standby updating and active restoring logic into ChangelogReader and extract out of StreamThread. guozhangwang#3 guozhangwang#4 Introduce Task state life cycle (created, restoring, running, suspended, closing), and refactor the task operations based on the current state. guozhangwang#6 guozhangwang#7 Consolidate AssignedTasks into TaskManager and simplify the logic of changelog management and task management (since they are already moved in step 2) and 3)). guozhangwang#8 guozhangwang#9 Also simplified the StreamThread logic a bit as the embedded clients / changelog restoration logic has been moved into step 1) and 2). guozhangwang#10 Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Bruno Cadonna <bruno@confluent.io>, Boyang Chen <boyang@confluent.io>

John Roesler and others added 5 commits January 10, 2020 14:11

added states for tasks and factored out cluster metadata from taskMan…

159c48b

…ager

wip

34b195c

rebase from John's branch

66cd779

resolve conflicts

cf4eb57

refactor standby tasks

401d169

vvcephei mentioned this pull request Jan 17, 2020

KAFKA-9441: Refactor Kafka Streams commit logic apache/kafka#7977

Closed

John Roesler and others added 5 commits January 17, 2020 09:57

wip

58f8a2a

oof. a lot of work in progress :/

8e2102c

wip

6b8e86c

unit tests phase one

8d8048a

rebase from John's branch

13ce59b

guozhangwang reviewed Jan 20, 2020

View reviewed changes

John Roesler and others added 9 commits January 21, 2020 15:59

fixing tests

ef84e91

rebase on John's branch

be24811

progress

89b0532

progress

9c0113d

fix

0b21b85

Merge branch 'refactor-task-lifecycle' of github.com:vvcephei/kafka i…

64b0ed5

…nto refactor-task-lifecycle

fix

364fe16

cleanup StandbyTaskTest

1c52f56

rebase again

edd963e

guozhangwang reviewed Jan 21, 2020

View reviewed changes

mjsax reviewed Jan 22, 2020

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java Outdated

guozhangwang reviewed Jan 22, 2020

View reviewed changes

John Roesler and others added 5 commits January 21, 2020 21:22

fix build

9bb38b5

fix build

e75a99f

fix build

efc5670

remove unused function

bbcf01f

Merge branch 'refactor-task-lifecycle' of https://github.com/vvcephei…

6ae87d1

…/kafka into K9113-standby-task-p5

guozhangwang mentioned this pull request Jan 22, 2020

KAFKA-9113: Unit test for RecordCollector #5

Merged

3 tasks

John Roesler and others added 19 commits January 24, 2020 12:33

review and added CLOSING state

d2804cf

cleanup

18b18cf

comment

581b909

review

7ebbd44

review

26d79f2

ignore more tests

f6818ee

rebase from John's branch

34e841e

disable heuristic

729b879

Merge branch 'refactor-task-lifecycle' of https://github.com/vvcephei…

89c8da3

…/kafka into K9113-standby-task-p5

fix SPAT

af971f1

fix topology test driver

6e92ec0

Merge branch 'refactor-task-lifecycle' of https://github.com/vvcephei…

9e03c60

…/kafka into K9113-standby-task-p5

fix more tests

4ee3efe

fixed more tests

3b0bfcc

formatting fix

4ef89c8

fix formatting

1f521e7

fix checkForCompletion

40add28

fix consumer pause / resume

d6b05e7

rebase

2574c23

guozhangwang reviewed Jan 24, 2020

View reviewed changes

guozhangwang changed the title ~~[WIP] Refactor task lifecycle~~ KAFKA-9113: P4, Refactor Task Lifecycle Jan 24, 2020

guozhangwang merged commit b3c8148 into guozhangwang:k9113-base Jan 24, 2020

vvcephei deleted the refactor-task-lifecycle branch January 25, 2020 00:52

vvcephei restored the refactor-task-lifecycle branch January 27, 2020 00:23

guozhangwang mentioned this pull request Jan 28, 2020

KAFKA-9113: Clean up task management and state management apache/kafka#7997

Merged

3 tasks

mjsax reviewed Jan 30, 2020

View reviewed changes

		CREATED, RESTORING, RUNNING, REVOKED, CLOSED;


		static void validateTransition(final State oldState, final State newState) {

Conversation

vvcephei commented Jan 15, 2020

Committer Checklist (excluded from commit message)

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!

Choose a reason for hiding this comment

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!

Uh oh!

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

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!

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!

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!

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!

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

guozhangwang Jan 21, 2020 •

edited

Loading

guozhangwang Feb 3, 2020 •

edited

Loading