KAFKA-10097: Internalize checkpoint data by abbccdda · Pull Request #8820 · apache/kafka

abbccdda · 2020-06-06T00:27:07Z

make checkpoint an internal data structure to avoid awkward passing around.
remove the possibility of getting a null checkpoint map

Committer Checklist (excluded from commit message)

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

mjsax · 2020-06-06T00:37:32Z

Do we need to set checkpointNeeded = false here?

mjsax · 2020-06-06T00:37:52Z

mjsax

Overall LGTM.

abbccdda · 2020-06-06T00:49:47Z

-    private Map<TopicPartition, Long> prepareClose(final boolean clean) {
-        final Map<TopicPartition, Long> checkpoint;
+    private void prepareClose(final boolean clean) {
+        // Reset any previously scheduled checkpoint.


I think we could just reset any checkpoint at the beginning, so that we only do checkpointing if this call thinks so. @mjsax

That should work.

mjsax · 2020-06-06T01:10:42Z

-        if (clean && checkpoint != null) {
-            executeAndMaybeSwallow(clean, () -> stateMgr.checkpoint(checkpoint), "state manager checkpoint", log);
+    private void close(final boolean clean) {
+        if (clean && checkpointNeeded) {


A if (clean) should be sufficient now

mjsax · 2020-06-06T01:44:34Z

Retest this please.

guozhangwang · 2020-06-06T02:30:09Z

test this

guozhangwang

Just one nit comment, otherwise LGTM!

guozhangwang · 2020-06-06T02:33:22Z

    private boolean commitNeeded = false;
    private boolean commitRequested = false;

+    private boolean checkpointNeeded = false;


Do we need two parameters here? Could we still just use one Map<TopicPartition, Long> checkpoint, and then in writeCheckpointIfNeed if it is not null write it and then set it to null.

I actually had a similar thought, but I am torn though. Using two variables required to keep them "in sync" was is not great. However, using null is less explicit... Thus overall I am fine either way as both seems to provide the overall same good/bad ratio.

A third option is Optional<Map<TopicPartition, Long>> which explicitly states the checkpoint is null. However, I think the benefit here is minimal, as checkpointNeeded sounds like a better state variable, than relying on the struct itself.

Actually I'm not too concerned of relying on null to indicate no need to checkpoint, what I originally pointed out is that we are unnecessarily accumulating and then distributing checkpoints between task manager and task, which is already resolved in your PR. So I'd suggest we still just use a single nullable Map.

guozhangwang · 2020-06-06T02:33:55Z

test this please

guozhangwang · 2020-06-06T02:34:21Z

test this

guozhangwang · 2020-06-06T15:50:39Z

test this

guozhangwang · 2020-06-06T15:51:36Z

okay to test

guozhangwang · 2020-06-06T15:52:01Z

okay to test

mjsax · 2020-06-06T20:13:11Z

Retest this please.

mjsax · 2020-06-06T20:13:25Z

Retest this please.

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax · 2020-06-07T00:44:21Z

Merged to trunk and cherry-picked to 2.6 branch.

Conflicts: * build.gradle: take upstream changes regarding heap memory configuration for the build. * apache-github/trunk: (33 commits) MINOR: fix HTML markup (apache#8823) KAFKA-10012; Reduce overhead of strings in SelectorMetrics (apache#8684) KAFKA-9216: Enforce internal config topic settings for Connect workers during startup (apache#8270) KAFKA-10097: Internalize checkpoint data (apache#8820) KAFKA-10033: Throw UnknownTopicOrPartitionException when modifying a non-existent topic's config MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) ...

mjsax added the streams label Jun 6, 2020

mjsax reviewed Jun 6, 2020

View reviewed changes

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

mjsax reviewed Jun 6, 2020

View reviewed changes

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

mjsax Jun 6, 2020

Copy link
Copy Markdown

Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As above?

mjsax reviewed Jun 6, 2020

View reviewed changes

internalize checkpoint

5b25a3a

abbccdda force-pushed the KAFKA-10097 branch from a823c53 to 5b25a3a Compare June 6, 2020 00:48

abbccdda commented Jun 6, 2020

View reviewed changes

mjsax reviewed Jun 6, 2020

View reviewed changes

minor change

6b7bcd1

guozhangwang reviewed Jun 6, 2020

View reviewed changes

only checkpoint struct

bc9c9da

mjsax merged commit d4ef46c into apache:trunk Jun 7, 2020

mjsax pushed a commit that referenced this pull request Jun 7, 2020

KAFKA-10097: Internalize checkpoint data (#8820)

e134bb4

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <guozhang@confluent.io>

Uh oh!

Conversation

abbccdda commented Jun 6, 2020

Committer Checklist (excluded from commit message)

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

mjsax commented Jun 6, 2020

Uh oh!

guozhangwang commented Jun 6, 2020

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!

guozhangwang commented Jun 6, 2020

Uh oh!

guozhangwang commented Jun 6, 2020

Uh oh!

guozhangwang commented Jun 6, 2020

Uh oh!

guozhangwang commented Jun 6, 2020

Uh oh!

guozhangwang commented Jun 6, 2020

Uh oh!

mjsax commented Jun 6, 2020

Uh oh!

mjsax commented Jun 6, 2020

Uh oh!

mjsax commented Jun 7, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants