KAFKA-9113: Clean up task management and state management by guozhangwang · Pull Request #7997 · apache/kafka

guozhangwang · 2020-01-22T05:46:31Z

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.
KAFKA-9113, P2: Extract Producer to RecordCollector guozhangwang/kafka#2
KAFKA-9113: Unit test for RecordCollector guozhangwang/kafka#5
Consolidate the standby updating and active restoring logic into ChangelogReader and extract out of StreamThread.
KAFKA-9113, P3: ProcessorStateManager and ChangelogReader guozhangwang/kafka#3
KAFKA-9113: Unit test for ProcessorStateManager and ChangelogReader guozhangwang/kafka#4
Introduce Task state life cycle (created, restoring, running, suspended, closing), and refactor the task operations based on the current state.
KAFKA-9113: P4, Refactor Task Lifecycle guozhangwang/kafka#6
KAFKA-9113: P5, StandbyTask Life Cycle guozhangwang/kafka#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)).
Fix TaskManagerTest guozhangwang/kafka#8
KAFKA-9113: StreamTask unit test guozhangwang/kafka#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

Committer Checklist (excluded from commit message)

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

…cord-collector

…ang/kafka into K9113-record-collector-p2

…ngwang/kafka into K9113-state-manager-p3

KAFKA-9113: Fix system / unit tests

guozhangwang · 2020-02-04T04:31:07Z

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

…to K9113-rebase

guozhangwang · 2020-02-04T20:48:40Z

@vvcephei This PR is ready for final reviews.

guozhangwang · 2020-02-04T20:51:02Z

retest this please

vvcephei

Hey @guozhangwang , Here's all the comments I have so far. I'm still in progress on the review, but figured I could submit these.

vvcephei · 2020-02-04T21:32:26Z

+                        log.debug("State store {} initialized from checkpoint with offset {} at changelog {}",
+                            store.stateStore.name(), store.offset, store.changelogPartition);
+                    } else {
+                        // TODO K9113: for EOS when there's no checkpointed offset, we should treat it as TaskCorrupted


unresolved TODO here. it's unclear to me why we would only consider this task corrupted in EOS mode. It seems like the lack of a checkpoint file just means that we loaded a cached task in an undefined state, and we should discard it and restore. If we have the file, but some of a changelog is missing from it, then maybe it just never got written, before shutdown, or more likely, the topology has changed and the task is corrupted, whether or not we are in EOS.

Today we write the checkpoint file before we commit the offsets, so under non-EOS we can just restore from scratch without wiping the local store image since restoring is just overwriting the local store and we can just restore to the end of the changelog (this may result in duplicates but is fine under non-EOS); but with non-EOS we have to first wipe out before restoring from scratch.

vvcephei · 2020-02-04T21:32:44Z

+            }
+
+            if (!loadedCheckpoints.isEmpty()) {
+                log.warn("Some loaded checkpoint offsets cannot find their corresponding state stores: {}", loadedCheckpoints);


seems like this also indicates the task is corrupted

Arguably yes, I'm just following the old logic intentionally here -- I think originally we want to be more compatible with topology changes (if some topology optimizations decides that some stores are not needed) but on second thought I think this is not safe either. We can consider making this change in another PR to make it stricter.

vvcephei · 2020-02-04T21:33:09Z

+                log.warn("Some loaded checkpoint offsets cannot find their corresponding state stores: {}", loadedCheckpoints);
+            }
+
+            checkpointFile.delete();


in retrospect, I like the idea of doing this regardless of EOS. Why should we deliberately produce wrong results in ALO mode? We can certainly optimize to be able to use semi-trustworthy data, but let's treat that separately from EOS vs ALO

The idea is that for ALO if we failed to write the first checkpoint after restarting we can still fallback to the original checkpoint even though the store may have been updated and hence there would be duplicates. But since the window gap (just one commit interval) is so small I think it does not worth making the code more complicated with EO v.s. ALO.

vvcephei · 2020-02-04T21:34:55Z

+        return taskId;
+    }
+
+    // used by the changelog reader only


these kinds of comments paradoxically lead to unmaintainability. Either a method is part of the public contract or not. If it is, then this comment would become out of date, if not, then the changelog reader shouldn't be using it. The recommendation is simply to delete the comment (and similar ones). This even applies to "for testing" comments. I have found several methods in use in this code base that were commented "visible for testing". Even for tests, either move both the class and the test into an isolated package and use package-private, refactor the test, or remove the comment. Also, if the changelog reader really needs four "holes" poked into this class, then we should reconsider the relationship between the state manager and the changelog reader.

vvcephei · 2020-02-04T21:37:40Z

-    private static Map<TopicPartition, Long> validCheckpointableOffsets(
-        final Map<TopicPartition, Long> checkpointableOffsets,
-        final Set<TopicPartition> validCheckpointableTopics) {
+    private StateStoreMetadata findStore(final TopicPartition changelogPartition) {


maybe it doesn't matter, but this method makes the only usage an n^2 algorithm. If we instead inverted the stores collection and used it for lookups in register, it would be o(n)

I've thought about it: inverting the stores collection makes other calls that depends on the store name more complicated, while keeping two collections indexed by storeName / changelog partition is not much worthy since within a task there are usually no more than 10 stores so this n^2 algorithm should not be a big deal.

…nals/TaskManager.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

vvcephei · 2020-02-04T22:42:56Z

-    private final byte[] recordValue = intSerializer.serialize(null, 10);
-    private final byte[] recordKey = intSerializer.serialize(null, 1);
+    @Mock(type = MockType.NICE)
+    private ChangelogReader changelogReader;


This is unused

vvcephei · 2020-02-04T22:43:06Z

+    private ProcessorStateManager stateManager;
+
+    @Mock(type = MockType.NICE)
+    private RecordCollector recordCollector;


This is also unused

vvcephei · 2020-02-04T22:49:18Z

+
+    @Ignore
    @Test
+    // FIXME: should unblock this test after we added invalid offset handling


Did you want to fix this as part of this PR or as a follow-on?

As a follow-up.

vvcephei

Hey @guozhangwang , I've completed my final review pass. I left a few comments earlier, but nothing that would stop me from merging this.

Thanks for driving this!

guozhangwang · 2020-02-04T23:10:50Z

https://jenkins.confluent.io/job/system-test-kafka-branch-builder/3732/ triggered another system test.

guozhangwang · 2020-02-04T23:53:21Z

-            if (generation() != Generation.NO_GENERATION) {
-                e = invokePartitionsRevoked(droppedPartitions);
-            } else {
+            if (generation() == Generation.NO_GENERATION || rebalanceInProgress()) {


@ableegoldman @mjsax This is another bug I found while trouble-shooting the system test failures (dates before the cleanup): when we got a task-migrated exception, and then enforce a rebalance, we call unsubscribe which would trigger onLeavePrepare, here if it was from task-migrated then it is likely that we are already undergoing a rebalance and in that case we should lose the tasks instead of revoke them since otherwise, we would still try to commit which would fail with a RebalanceInProgress exception.

Good catch. Also it's pretty unfortunate that we can only trigger a rebalance from outside the client by unsubscribing and closing/suspending the entire assignment...this limits the usefulness of KIP-429 during version probing upgrades.

It also has some implications for the "rebalances are cheap" assumption of KIP-441. Would be better phrased as "rebalances are cheap, except for the member who triggers them".

Yeah ... maybe we should consider adding a new API to consumer to rejoin the group, in a cheaper way.

I agree :) -- would be happy to write up a small KIP for it and kick off discussion

guozhangwang · 2020-02-05T04:58:42Z

https://jenkins.confluent.io/job/system-test-kafka-branch-builder/3732/console succeed, merging to trunk now.

…to k9113-base

Conflicts: * build.gradle: moved avro plugin definition below newly added test retry plugin. * apache-github/trunk: MINOR: further InternalTopologyBuilder cleanup (apache#8046) MINOR: Add timer for update limit offsets (apache#8047) HOTFIX: Fix spotsbug failure in Kafka examples (apache#8051) KAFKA-9447: Add new customized EOS model example (apache#8031) KAFKA-8164: Add support for retrying failed (apache#8019) HOTFIX: checkstyle for newly added unit test KAFKA-9261; Client should handle unavailable leader metadata (apache#7770) MINOR: Fix typos introduced in KIP-559 (apache#8042) MINOR: Fixing null handilg in ValueAndTimestampSerializer (apache#7679) KAFKA-9113: Clean up task management and state management (apache#7997) MINOR: fix checkstyle issue in ConsumerConfig.java (apache#8038) KAFKA-9491; Increment high watermark after full log truncation (apache#8037) KAFKA-9477 Document RoundRobinAssignor as an option for partition.assignment.strategy (apache#8007) KAFKA-9074: Correct Connect’s `Values.parseString` to properly parse a time and timestamp literal (apache#7568) KAFKA-9492; Ignore record errors in ProduceResponse for older versions (apache#8030)

…t-for-generated-requests * apache-github/trunk: (410 commits) KAFKA-8843: KIP-515: Zookeeper TLS support MINOR: Add missing quote for malformed line content (apache#8070) MINOR: Simplify KafkaProducerTest (apache#8044) KAFKA-9507; AdminClient should check for missing committed offsets (apache#8057) KAFKA-9519: Deprecate the --zookeeper flag in ConfigCommand (apache#8056) KAFKA-9509; Fixing flakiness of MirrorConnectorsIntegrationTest.testReplication (apache#8048) HOTFIX: Fix two test failures in JDK11 (apache#8063) DOCS - clarify transactionalID and idempotent behavior (apache#7821) MINOR: further InternalTopologyBuilder cleanup (apache#8046) MINOR: Add timer for update limit offsets (apache#8047) HOTFIX: Fix spotsbug failure in Kafka examples (apache#8051) KAFKA-9447: Add new customized EOS model example (apache#8031) KAFKA-8164: Add support for retrying failed (apache#8019) HOTFIX: checkstyle for newly added unit test KAFKA-9261; Client should handle unavailable leader metadata (apache#7770) MINOR: Fix typos introduced in KIP-559 (apache#8042) MINOR: Fixing null handilg in ValueAndTimestampSerializer (apache#7679) KAFKA-9113: Clean up task management and state management (apache#7997) MINOR: fix checkstyle issue in ConsumerConfig.java (apache#8038) KAFKA-9491; Increment high watermark after full log truncation (apache#8037) ...

guozhangwang added 30 commits December 13, 2019 20:09

record collector inheritance cleanup

067cf6e

unit tests

76b2731

extract producer out of stream task

ff672ef

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

7c1f06e

…cord-collector

github comments

c47a66c

further cleanup

9d9c7da

Merge branch 'K9113-record-collector' of https://github.com/guozhangw…

2f67f35

…ang/kafka into K9113-record-collector-p2

fix unit tests

38be1c1

fix more unit tests

d4290d4

more unit tests

2aa6673

rebased from trunk

edd0265

fix standby task tests

578d34c

minor fix

412d69d

more unit tests

b94579f

refactor initialization

7e632fc

one last unit test

347ea26

refactor changelog reader

5108872

initialize and check for metadata availability

ca9a9aa

minor

4007d2a

minor

45a9a44

delete state restorer

191546c

standby task

eecc485

changelog update limit offsets

fd77385

refactor changelog reader

9eae4a6

address comments

7d1cef5

Merge branch 'K9113-record-collector-p2' of https://github.com/guozha…

b2ed341

…ngwang/kafka into K9113-state-manager-p3

remove StateStoreException javadoc

c376017

address comments

995979a

update some TODO K9113

f1aafe2

Merge branch 'K9113-record-collector-p2' of https://github.com/guozha…

9ba5029

…ngwang/kafka into K9113-state-manager-p3

guozhangwang added 3 commits February 3, 2020 17:52

rebase from trunk

f5b1730

rebase from base

bd79a51

Merge pull request #11 from guozhangwang/K9113-rebase

d894ac7

KAFKA-9113: Fix system / unit tests

guozhangwang added 4 commits February 3, 2020 22:01

remove the unnecessary shortcut

8c6746b

check for rebalance-in-progress onLeavePrepare as well

a6ab6d5

Merge branch 'k9113-base' of https://github.com/guozhangwang/kafka in…

1c04aa3

…to K9113-rebase

fix findbugs

e3a38ab

Merge branch 'trunk' of https://github.com/apache/kafka into k9113-base

9d0d3c2

vvcephei reviewed Feb 4, 2020

View reviewed changes

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

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

3e20571

…nals/TaskManager.java Co-Authored-By: John Roesler <vvcephei@users.noreply.github.com>

vvcephei reviewed Feb 4, 2020

View reviewed changes

vvcephei approved these changes Feb 4, 2020

View reviewed changes

guozhangwang commented Feb 4, 2020

View reviewed changes

guozhangwang added 2 commits February 4, 2020 20:59

address comments

3db079a

Merge branch 'k9113-base' of https://github.com/guozhangwang/kafka in…

196e7c0

…to k9113-base

guozhangwang merged commit 4090f9a into apache:trunk Feb 5, 2020

vvcephei mentioned this pull request Mar 2, 2020

MINOR: Port streams broker compatibility fix #8203

Merged

3 tasks

guozhangwang deleted the k9113-base branch April 24, 2020 23:59

chia7712 mentioned this pull request Feb 20, 2021

MINOR: remove unused org.apache.kafka.streams.processor.internals.Res… #10164

Merged

3 tasks

Conversation

guozhangwang commented Jan 22, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

guozhangwang commented Feb 4, 2020

Uh oh!

guozhangwang commented Feb 4, 2020

Uh oh!

guozhangwang commented Feb 4, 2020

Uh oh!

vvcephei 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!

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!

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 4, 2020

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 Feb 5, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

guozhangwang commented Jan 22, 2020 •

edited

Loading