KAFKA-10199: Fix restoration behavior for paused tasks #14437

lucasbru · 2023-09-25T08:11:01Z

State updater can get into a busy loop when all tasks are paused, because changelogReader will never return that all changelogs have been read completely. Fix this, by awaiting if updatingTasks is empty.

Related and included: if we are restoring and all tasks are paused, we should return immediately from StoreChangelogReader.

Committer Checklist (excluded from commit message)

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

lucasbru · 2023-09-25T08:12:57Z

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

@@ -1043,6 +1046,28 @@ public void shouldResumeActiveStatefulTask() throws Exception {
        verify(changelogReader, times(2)).enforceRestoreActive();
    }

+    @Test
+    public void shouldAwaitWhenAllTasksPaused() throws Exception {


This tests the absence of "restore" calls when all tasks are paused. This requires "smelly" constructs like Mockito.reset and Thread.sleep. We could alternatively not test this - the absence of something in a certain timeframe is not very amenable to unit testing.

cadonna

Thanks for the PR, @lucasbru !

Here my feedback!

cadonna · 2023-09-25T09:24:32Z

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

+
+        verifyPausedTasks(task);
+
+        reset(changelogReader);


I think, you do not need to reset the mock since we guarantee that paused tasks that are added to the state updater are not restored. We miss a test for that case, though.

I don't think this is quite true. Paused tasks will not be added to updatingTasks, but we may still go around the loop calling restore on a potentially empty set. This is testing that after a while, we do not go around the loop anymore. But it's actually not a good test, for the reason described above.

cadonna · 2023-09-25T10:00:25Z

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

+        Thread.sleep(100);
+        verify(changelogReader, never()).restore(any());


Instead of this, we could expose the thread state on the implementation of the state updater, but not on the interface. With that and a wait until verification, we could verify that the thread changes to the waiting state. We similar with getActiveTasks() or getPausedTasks(). That does also smell a bit, but it avoids the sleep and maybe smells a little bit less? 🙂

I like the idea!

cadonna · 2023-09-25T10:00:43Z

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

@@ -307,7 +307,7 @@ private void waitIfAllChangelogsCompletelyRead() {
            tasksAndActionsLock.lock();
            try {
                while (isRunning.get() &&
-                    changelogReader.allChangelogsCompleted() &&
+                    (changelogReader.allChangelogsCompleted() || updatingTasks.isEmpty()) &&


Nice catch!

- If all tasks are paused, we should wait in the state updater - If all tasks are paused, we should return immediately from `StoreChangelogReader`. For testing, expose an `isIdle` state to check that we enter an idling state eventually.

…auses

cadonna

LGTM!

Thanks @lucasbru !

cadonna · 2023-09-26T12:06:10Z

Build failures are unrelated.

* KAFKA-10199: Fix restoration behavior for paused tasks (apache#14437) State updater can get into a busy loop when all tasks are paused, because changelogReader will never return that all changelogs have been read completely. Fix this, by awaiting if updatingTasks is empty. Related and included: if we are restoring and all tasks are paused, we should return immediately from StoreChangelogReader. Reviewer: Bruno Cadonna <cadonna@apache.org> * KAFKA-10199: Do not process when in PARTITIONS_REVOKED (apache#14265) When a Streams application is subscribed with a pattern to input topics and an input topic is deleted, the stream thread transists to PARTITIONS_REVOKED and a rebalance is triggered. This happens inside the poll call. Sometimes, the poll call returns before a new assignment is received. That means, Streams executes the poll loop in state PARTITIONS_REVOKED. With the state updater enabled processing is also executed in states other than RUNNING and so processing is also executed when the stream thread is in state PARTITION_REVOKED. However, that triggers an IllegalStateException with error message: No current assignment for partition TEST-TOPIC-A-0 which is a fatal error. This commit prevents processing when the stream thread is in state PARTITIONS_REVOKED. Reviewer: Lucas Brutschy <lbrutschy@confluent.io> * KAFKA-15326: [8/N] Move consumer interaction out of processing methods (apache#14226) The process method inside the tasks needs to be called from within the processing threads. However, it currently interacts with the consumer in two ways: * It resumes processing when the PartitionGroup buffers are empty * It fetches the lag from the consumer We introduce updateLags() and resumePollingForPartitionsWithAvailableSpace() methods that call into the task from the polling thread, in order to set up the consumer correctly for the next poll, and extract metadata from the consumer after the poll. Reviewer: Bruno Cadonna <bruno@confluent.io> --------- Co-authored-by: Lucas Brutschy <lbrutschy@confluent.io> Co-authored-by: Bruno Cadonna <cadonna@apache.org>

State updater can get into a busy loop when all tasks are paused, because changelogReader will never return that all changelogs have been read completely. Fix this, by awaiting if updatingTasks is empty. Related and included: if we are restoring and all tasks are paused, we should return immediately from StoreChangelogReader. Reviewer: Bruno Cadonna <cadonna@apache.org>

lucasbru commented Sep 25, 2023

View reviewed changes

lucasbru requested a review from cadonna September 25, 2023 08:13

lucasbru added the streams label Sep 25, 2023

cadonna reviewed Sep 25, 2023

View reviewed changes

lucasbru added 2 commits September 26, 2023 09:56

Merge remote-tracking branch 'upstream/trunk' into ak_state_updater_p…

5b828ba

…auses

lucasbru force-pushed the ak_state_updater_pauses branch from 504e123 to 5b828ba Compare September 26, 2023 07:58

lucasbru added 2 commits September 26, 2023 10:02

fix

aeb9f09

fix

e6f0d05

cadonna approved these changes Sep 26, 2023

View reviewed changes

fix

8c3edfa

cadonna merged commit 2d04370 into apache:trunk Sep 26, 2023
1 check was pending

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-10199: Fix restoration behavior for paused tasks #14437

KAFKA-10199: Fix restoration behavior for paused tasks #14437

lucasbru commented Sep 25, 2023

lucasbru Sep 25, 2023

cadonna left a comment

cadonna Sep 25, 2023

lucasbru Sep 26, 2023

cadonna Sep 25, 2023

lucasbru Sep 26, 2023

cadonna Sep 25, 2023

cadonna left a comment

cadonna commented Sep 26, 2023

		Thread.sleep(100);
		verify(changelogReader, never()).restore(any());

KAFKA-10199: Fix restoration behavior for paused tasks #14437

KAFKA-10199: Fix restoration behavior for paused tasks #14437

Conversation

lucasbru commented Sep 25, 2023

Committer Checklist (excluded from commit message)

lucasbru Sep 25, 2023

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

cadonna Sep 25, 2023

Choose a reason for hiding this comment

lucasbru Sep 26, 2023

Choose a reason for hiding this comment

cadonna Sep 25, 2023

Choose a reason for hiding this comment

lucasbru Sep 26, 2023

Choose a reason for hiding this comment

cadonna Sep 25, 2023

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

cadonna commented Sep 26, 2023