KAFKA-10199: Commit the restoration progress within StateUpdater #12279

guozhangwang · 2022-06-10T01:20:58Z

During restoring, we should always commit a.k.a. write checkpoint file regardless of EOS or ALOS, since if there's a failure we would just over-restore them upon recovery so no EOS violations happened.

Also when we complete restore or remove task, we should enforce a checkpoint as well; for failing cases though, we should not write a new one.

Committer Checklist (excluded from commit message)

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

…ommit-in-state-updater

cadonna

@guozhangwang Thanks for the PR!

Here my comment.

cadonna · 2022-06-16T14:31:31Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

+                }
+
+                for (final Task task : updatingTasks.values()) {
+                    // do not enforce checkpointing during restoration if its position has not advanced much


Is this a ToDo?

This is not a Todo: it's explaining why we set the enforceCheckpoint as false. Inside that callee when it's false we will only write a new checkpoint if the offsets has significantly advanced.

I see! This is a bit hard to understand in my opinion. Could we have two methods -- commitTaskAndEnforceCheckpoint() and commitTaskAndMaybeEnforcedCheckpoint()? If we change the code to only write the checkpoints, this code might change anyways.

Fair enough. I will refactor this piece of code in the follow-up PR (I have one doing the code refactoring already) such that:

We remove the prepareCommit and postCommit in standby task, instead we call writeCheckpoint directly.

We remove the postCommit in active task, enforceCheckpoint boolean in the maybeCheckpoint function.

Looking forward to the follow-up PR 🙂

cadonna · 2022-06-16T14:35:11Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

+                throw new IllegalStateException("Task " + task.id() + " should not have any source offset " +
+                        "committable during restoration, but have " + offsetAndMetadata + " instead. " + BUG_ERROR_MESSAGE);
+            }
+


Suggested change

cadonna · 2022-06-16T14:36:38Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java


    @AfterEach
    public void tearDown() {
        stateUpdater.shutdown(Duration.ofMinutes(1));
    }

+    private Properties configProps(final int commitInterval) {
+        return mkObjectProperties(mkMap(
+                mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, safeUniqueClassTestName(getClass())),


Why do we need this since we do not use an embedded Kafka cluster?

These two configs are required configs of StreamsConfig and hence we have to provide a dummy.

Yes, I know. My question was actually about if we need to use safeUniqueClassTestName(getClass()) and provide a new override for that method. AFAIK, that method is primarily used in integration tests to ensure that consumer groups within one integration test class have distinct names so that consumer groups of different tests in the test class do not clash leading to longer execution times. This is not an integration test and we do not use an embedded Kafka cluster. Why do we need safeUniqueClassTestName(getClass())?

That's fair. I will remove the safeUniqueClassTestName and just dummy with appID.

cadonna · 2022-06-16T14:37:14Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+                mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2171"),
+                mkEntry(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2),
+                mkEntry(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, commitInterval),
+                // we need to make sure that transaction timeout is not lower than commit interval for EOS


Is this also a ToDo?

Not a todo: I was explaining why we need to explicitly set the RANSACTION_TIMEOUT_CONFIG as commitInterval as well.

Do we have a check for this somewhere in our production code?

Yes, otherwise the app would fail upon starting up.

If there is a check in the production code, we do not need to add this comment since the test would fail anyways, right? I would remove the comment.

cadonna · 2022-06-16T14:38:20Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+            verify(task, times(0)).prepareCommit();
+            verify(task, times(0)).postCommit(true);
+            verify(task, times(0)).postCommit(false);


Suggested change

verify(task, times(0)).prepareCommit();

verify(task, times(0)).postCommit(true);

verify(task, times(0)).postCommit(false);

verify(task, never()).prepareCommit();

verify(task, never()).postCommit(true);

verify(task, never()).postCommit(false);

cadonna · 2022-06-16T14:59:51Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+        waitForCondition(
+                () -> {
+                    for (final Task task : tasks) {
+                        verify(task, atLeast(1)).prepareCommit();
+                        verify(task, atLeast(1)).postCommit(enforceCheckpoint);
+                    }
+
+                    return true;
+                },
+                VERIFICATION_TIMEOUT,
+                "Did not auto commit all tasks within the given timeout!"
+        );


Did you consider to use verify(task, timeout(VERIFICATION_TIMEOUT).atLeast(1)) instead of waitForCondition()?

Thanks for the tip! will give it a try.

cadonna · 2022-06-16T15:02:20Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+
+                    return true;
+                },
+                VERIFICATION_TIMEOUT,


Could you express the timeout in terms of the commit interval?

I tried it a bit but since we are using a system test here using exact commit interval could cause flakiness; while using a larger value say commit interval * 2 is safe enough I felt it would be similar to just using a longer value as VERIFICATION_TIMEOUT here.

cadonna · 2022-06-16T15:09:50Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

+
+        private void commitTask(final Task task, final boolean enforceCheckpoint) {
+            // prepare commit should not take any effect except a no-op verification
+            final Map<TopicPartition, OffsetAndMetadata> offsetAndMetadata = task.prepareCommit();


I am not sure I understand why we should commit. The task does not read from the input at this point. Wouldn't flushing the stores and writing the checkpoint file be enough? Can we somehow just flush the state store and write the checkpoint instead of calling the *Commit() methods? I think that would simplify the code.

I was trying to leverage on the extra state validation logic inside the *Commit() function, but I agree we can directly call checkpointing which would be more straight-forward. LMK if you feel strong about this and I can change it accordingly.

Yes, I feel strongly about it. 🙂

I plan to do the function refactoring (mentioned above) in the follow-up PR, and here I would just directly call the checkpoint functions.

cadonna · 2022-06-16T15:17:11Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -286,6 +308,7 @@ public void shouldRestoreActiveStatefulTasksAndUpdateStandbyTasks() throws Excep
        stateUpdater.add(task4);

        verifyRestoredActiveTasks(task2, task1);
+        verifyCommitTasks(true, task2, task1);


Shouldn't the offsets of standby tasks also be written to the checkpoint file?

We can only be certain that active tasks would be checkpointed since they are completed upon when we enforce the checkpoint; standby tasks would only try checkpointing without enforcing during processing and hence here they will not write the checkpoint.

What if you move this verification after verifyUpdatingStandbyTasks()? Are standby tasks still not checkpointed because we do not enforce the checkpoint?

Yes, unless we wait for commit interval.. btw I think since this is already covered in other tests we do not need to wait (and since it's system time we need to wait conservatively to reduce flakiness) again in this test.

…ommit-in-state-updater

cadonna

Thanks for the updates, @guozhangwang !

Here my feedback!

cadonna · 2022-06-20T08:30:40Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

@@ -85,7 +86,7 @@ public Collection<StandbyTask> getUpdatingStandbyTasks() {
        }

        public boolean onlyStandbyTasksLeft() {
-            return !updatingTasks.isEmpty() && updatingTasks.values().stream().allMatch(t -> !t.isActive());
+            return !updatingTasks.isEmpty() && updatingTasks.values().stream().noneMatch(Task::isActive);


I did the same change in #12312 🙂
I will revert the change in my PR to avoid merge conflicts.

cadonna · 2022-06-20T08:35:50Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -88,7 +88,8 @@ public abstract class AbstractTask implements Task {
     * @throws StreamsException fatal error when flushing the state store, for example sending changelog records failed
     *                          or flushing state store get IO errors; such error should cause the thread to die
     */
-    protected void maybeWriteCheckpoint(final boolean enforceCheckpoint) {
+    @Override
+    public void maybeCheckpoint(final boolean enforceCheckpoint) {


Now that this method is public, could you please add unit tests for this method?

Yes, will do.

cadonna · 2022-06-20T08:51:03Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

+                }
+
+                for (final Task task : updatingTasks.values()) {
+                    // do not enforce checkpointing during restoration if its position has not advanced much


Looking forward to the follow-up PR 🙂

cadonna · 2022-06-20T08:56:15Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+                mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2171"),
+                mkEntry(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2),
+                mkEntry(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, commitInterval),
+                // we need to make sure that transaction timeout is not lower than commit interval for EOS


If there is a check in the production code, we do not need to add this comment since the test would fail anyways, right? I would remove the comment.

cadonna · 2022-06-20T08:57:08Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java

+    public static String safeUniqueClassTestName(final Class<?> testClass) {
+        return (testClass.getSimpleName())
+                .replace(':', '_')
+                .replace('.', '_')
+                .replace('[', '_')
+                .replace(']', '_')
+                .replace(' ', '_')
+                .replace('=', '_');
+    }
+


This should be dead code now that we do not need this method in DefaultStateUpdater. Could you please remove it?

cadonna · 2022-06-20T10:50:32Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -506,6 +537,7 @@ public void shouldAddFailedTasksToQueueWhenRestoreThrowsStreamsExceptionWithTask
        verifyUpdatingTasks(task2);
        verifyRestoredActiveTasks();
        verifyRemovedTasks();
+        verifyNeverCheckpointTasks(task1, task3);


Same flakiness as above

cadonna · 2022-06-20T10:50:55Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -531,6 +563,7 @@ public void shouldAddFailedTasksToQueueWhenRestoreThrowsTaskCorruptedException()
        verifyUpdatingTasks(task3);
        verifyRestoredActiveTasks();
        verifyRemovedTasks();
+        verifyNeverCheckpointTasks(task1, task2);


Same flakiness as above

cadonna · 2022-06-20T10:51:16Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -552,6 +585,7 @@ public void shouldAddFailedTasksToQueueWhenUncaughtExceptionIsThrown() throws Ex
        verifyUpdatingTasks();
        verifyRestoredActiveTasks();
        verifyRemovedTasks();
+        verifyNeverCheckpointTasks(task1, task2);


Same flakiness as above

cadonna · 2022-06-20T11:02:16Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+        final StreamsConfig config = new StreamsConfig(configProps(Integer.MAX_VALUE));
+        final DefaultStateUpdater stateUpdater = new DefaultStateUpdater(config, changelogReader, offsetResetter, Time.SYSTEM);


It is better to do this, otherwise verifications like verifyExceptionsAndFailedTasks() will not work. You do not use them in this method, but the change makes the test future-proof.

Suggested change

final StreamsConfig config = new StreamsConfig(configProps(Integer.MAX_VALUE));

final DefaultStateUpdater stateUpdater = new DefaultStateUpdater(config, changelogReader, offsetResetter, Time.SYSTEM);

stateUpdater.shutdown(Duration.ofMinutes(1));

final StreamsConfig config = new StreamsConfig(configProps(Integer.MAX_VALUE));

stateUpdater = new DefaultStateUpdater(config, changelogReader, offsetResetter, Time.SYSTEM);

Sounds good, I will remove final on that variable then.

cadonna · 2022-06-20T13:15:43Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+
+    private void verifyCheckpointTasks(final boolean enforceCheckpoint, final Task... tasks) throws Exception {
+        for (final Task task : tasks) {
+            verify(task, timeout(VERIFICATION_TIMEOUT).atLeast(1)).maybeCheckpoint(enforceCheckpoint);


nit: I think the timeout is not needed here, since you use this verification after the tasks are either restored or removed. In those cases maybeCheckpoint() should have already be called by then.

When I removed the timeout(VERIFICATION_TIMEOUT) the shouldAutoCheckpointTasksOnInterval starts to be flaky (once every ~30 on my laptop). This is because in that test we do not close the task before verification, and the system time makes sleeping just one commit interval not sufficient to be 100% sure (even with commit interval * 2 I can still see flakiness). So I reverted that removal by the end.

…ommit-in-state-updater

guozhangwang · 2022-06-21T19:17:29Z

Thanks for having another look @cadonna , your comments are addressed.

guozhangwang · 2022-06-21T19:18:22Z

Oh I had one commit for adding the unit test lost, adding now.

cadonna

@guozhangwang Thanks for the updates

Here my feedback.

Feel free to react on my feedback in a follow-up PR.

cadonna · 2022-06-23T15:51:56Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java

+        task.maybeCheckpoint(false);  // this should not checkpoint
+        task.maybeCheckpoint(false);  // this should checkpoint
+        task.maybeCheckpoint(false);  // this should not checkpoint


This test is not really clear about when the checkpointing happens. Ideally, we would need to verify after each call to maybeCheckpoint(). That might be possible with EasyMock#reset() but I am not 100% sure.

Ack. I figured out a way without relying on reset(), and instead just checking on offsetSnapshotSinceLastFlush which is only updated if checkpoint is indeed executed.

cadonna · 2022-06-23T15:53:03Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java

+        task = createOptimizedStatefulTask(createConfig("100"), consumer);
+
+        task.initializeIfNeeded();
+        task.maybeCheckpoint(true);


Do we not need to also test task.maybeCheckpoint(false)?

Ack, added.

cadonna · 2022-06-23T15:59:51Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

@@ -313,6 +337,7 @@ public void shouldRestoreActiveStatefulTaskThenUpdateStandbyTaskAndAgainRestoreA
        stateUpdater.add(task2);

        verifyRestoredActiveTasks(task1);
+        verifyCheckpointTasks(true, task1);


I think we need another verification before line 350.

guozhangwang · 2022-06-23T17:44:46Z

Thanks @cadonna . I will follow-up on your comments for unit tests in the next PR.

cadonna · 2022-06-24T11:55:50Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java

+        stateUpdater.add(task3);
+        stateUpdater.add(task4);
+
+        sleep(COMMIT_INTERVAL);


Shall we use MockTime() instead of system time to better control the progress of time?

Sure: #12344

…che#12279) During restoring, we should always commit a.k.a. write checkpoint file regardless of EOS or ALOS, since if there's a failure we would just over-restore them upon recovery so no EOS violations happened. Also when we complete restore or remove task, we should enforce a checkpoint as well; for failing cases though, we should not write a new one. Reviewers: Bruno Cadonna <cadonna@apache.org>

guozhangwang added 2 commits June 9, 2022 18:16

commit during restoration

451288b

more unit tests

7d4f1cb

guozhangwang requested a review from cadonna June 10, 2022 01:23

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

9bfd5d1

…ommit-in-state-updater

cadonna reviewed Jun 16, 2022

View reviewed changes

guozhangwang added 4 commits June 16, 2022 08:45

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

0c04b51

…ommit-in-state-updater

github comments

b7a3ad7

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

9033c67

…ommit-in-state-updater

github comments

2e815dd

cadonna reviewed Jun 20, 2022

View reviewed changes

guozhangwang added 2 commits June 21, 2022 10:16

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

bb7b155

…ommit-in-state-updater

github comments

ebc29d6

guozhangwang added 2 commits June 21, 2022 14:37

unit test for maybeCheckpoint

e86524f

remove redundant test

afa8ccb

cadonna approved these changes Jun 23, 2022

View reviewed changes

guozhangwang merged commit 925c628 into apache:trunk Jun 23, 2022

guozhangwang mentioned this pull request Jun 23, 2022

KAFKA-10199: Remove main consumer from store changelog reader #12337

Merged

3 tasks

cadonna reviewed Jun 24, 2022

View reviewed changes

		final StreamsConfig config = new StreamsConfig(configProps(Integer.MAX_VALUE));
		final DefaultStateUpdater stateUpdater = new DefaultStateUpdater(config, changelogReader, offsetResetter, Time.SYSTEM);

KAFKA-10199: Commit the restoration progress within StateUpdater #12279

KAFKA-10199: Commit the restoration progress within StateUpdater #12279

Conversation

guozhangwang commented Jun 10, 2022

Committer Checklist (excluded from commit message)

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 17, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 17, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 17, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 17, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 20, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jun 21, 2022

guozhangwang commented Jun 21, 2022

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jun 23, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna Jun 17, 2022 •

edited

cadonna Jun 17, 2022 •

edited

cadonna Jun 17, 2022 •

edited

cadonna Jun 17, 2022 •

edited

cadonna Jun 20, 2022 •

edited