KAFKA-14299: Fix pause and resume with state updater by lucasbru · Pull Request #13025 · apache/kafka

lucasbru · 2022-12-20T16:29:17Z

Fixes required to make the PauseResumeIntegrationTest pass. It was not
enabled and it does not pass for the state updater code path.

Make sure no progress is made on paused topologies
The state updater restored one round of polls from the restore
consumer before realizing that a newly added task was already
in paused state when being added.
Wake up state updater when tasks are being resumed
If a task is resumed, it may be necessary to wake up the state
updater from waiting on the tasksAndActions condition.
Make sure that allTasks methods also return the tasks
that are currently being restored.
Enable PauseResumeIntegrationTest and upgrade it to
JUnit5.

Committer Checklist (excluded from commit message)

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

lucasbru · 2022-12-20T16:29:38Z

@cadonna Could you have a look?

cadonna

Thanks for the PR, @lucasbru !

Here my feedback!

cadonna · 2022-12-20T17:06:02Z

Why not have this if on line 264 and adding it to the pausedTasks right away instead of calling pauseTask() which does a couple of steps like writing a checkpoint which is not needed at when the task is just added to the state updater?

Done. But I also had to make more changes to make resuming reliable. I have to wake up the state updater thread when a topology is resumed, to make sure that the state updater does in fact resume the task

This wasn't enabled and didn't pass for state updater.

The state updater restored one round of polls from the restore consumer before realizing that a newly added task was already in paused state when being added. This is tested by the existing `PauseResumeIntegrationTest`.

cadonna

@lucasbru Thanks for the updates!

Here my feedback!

cadonna · 2023-01-12T13:00:18Z

-                log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+                log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+            } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+                pausedTasks.put(taskId, task);


To be consistent, we should also verify here if the task already exists similar to the else-branch.

I'm wondering if this complexity is necessary, since we do not make strict ordering guarantees for paused topologies -- i.e. it's okay to still processing those tasks for a while after the pause() call is triggered. Is it really a correctness or concurrency issue?

The PauseResumeIntegrationTest is testing precisely this (that no progress is made when a task is added in paused state). I also wasn't sure how important this property is for KSQL, but I suppose it is important enough that Jim wrote a unit test for it. I also think it makes sense to have the property, making a little progress on a paused task is unintuitive IMHO.

w.r.t. Bruno's comment: done

While working on another PR I realized in checkAllUpdatingTaskStates we only try to pause tasks when the commit interval has elapsed, i.e. when we pause a named topology, its corresponding tasks may only be paused after a while, while when we resume, tasks are resumed immediately.

So I think we should move the pausing logic out of the checkAllUpdatingTaskStates as well like resuming, which would leave checkAllUpdatingTaskStates to just become maybeCheckpointUpdatingTasks. WDYT?

Sorry, almost missed this comment. I think this would work, but it would mean cycling through the list of all tasks in every single iteration of main state updater loop. I thought the pause code was only run after commitMs to reduce this overhead, so I thought it makes sense. It's different for resume, because we get a resume signal from the stream thread.

cadonna · 2023-01-12T13:03:11Z

    private final Condition restoredActiveTasksCondition = restoredActiveTasksLock.newCondition();
    private final BlockingQueue<ExceptionAndTasks> exceptionsAndFailedTasks = new LinkedBlockingQueue<>();
    private final BlockingQueue<Task> removedTasks = new LinkedBlockingQueue<>();
+    private final AtomicBoolean needsResumeCheck = new AtomicBoolean(false);


nit: I find the name a bit confusing. What do you think about isTopologyResumed or 'topologyResumed'?

cadonna · 2023-01-12T13:13:45Z

+        if (stateUpdater != null) {
+            final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+            ret.putAll(tasks.allTasksPerId());
+            return ret;
+        } else {
+            return tasks.allTasksPerId();
+        }


I could not find a test for this change. Could you add one?

lucasbru · 2023-01-31T17:37:42Z

@guozhangwang also wanted to have a look (note that I didn't address Brunos comments yet)

guozhangwang · 2023-01-31T17:48:18Z

ack, will take a look asap.

guozhangwang · 2023-01-31T23:39:53Z

    @Override
    public boolean commitNeeded() {
-        throw new UnsupportedOperationException("This task is read-only");
+        return task.commitNeeded();


Why do we need to change these two functions?

StreamThread.maybeCommit uses TaskManager.allTasks that excluded tasks in the state updater before, but includes tasks in the state updater now. However, commitNeeded should be false for all those tasks right? I was scratching my head here a bit, because StreamThread.maybeCommit was processing RESTORING tasks in the old code path, but not in the state updater code path.

Ack. That makes sense.

It gives me a bit of a headache to forward the call to commitNeeded() since it changes the state of a stream task by updating the consumed offsets.
I am also wondering why Streams also commits restoring tasks.

I think the actual culprit is like @lucasbru said, StreamThread.maybeCommit uses TaskManager.allTasks, while the callee TaskManager.allTasks is used in many different places.. For example for IQ, it was used and in that case it should return all tasks, whereas in this caller, TaskManager.allTasks should just return all processing tasks when updater is enabled, i.e. only the ones in TaskRegistry.

I am also wondering why Streams also commits restoring tasks.

For that, the rationale is to allow the restoration progress to be recorded as well in case it was paused due to another rebalance (though it's prepareCommit would always return empty map, so the committing process for those restoring tasks would be reduced to writing checkpoints). But somehow in the middle of history we lost the logic to ever change commitNeeded flag for restoring tasks, so they would always not be triggered. Hence a new JIRA is proposed to fix it, and we fixed it in state updater.

I thought about fixing it in a probably better way, which involves more changes: 1) introduce a TaskManager.allProcessingTasks, 2) depending on the updater enabled flag, let maybeCommit call either this one or allTasks. When I thought about that, I admit I was thinking this change is a bit too much, and since tasks in updater the task should not make any side-effects and return commitNeeded false this maybe just okay.

But thinking about this a bit more, I think it's indeed worrisome to let ReadyOnlyTask.commitNeeded to have any potential side effects if we ever have bugs..

introduce a TaskManager.allProcessingTasks, 2) depending on the updater enabled flag, let maybeCommit call either this one or allTasks

let's do :this:
I will update the PR tomorrow first thing

guozhangwang · 2023-01-31T23:46:31Z

-                log.info("Stateless active task " + task.id() + " was added to the restored tasks of the state updater");
+                log.info("Stateless active task " + taskId + " was added to the restored tasks of the state updater");
+            } else if (topologyMetadata.isPaused(taskId.topologyName())) {
+                pausedTasks.put(taskId, task);


I'm wondering if this complexity is necessary, since we do not make strict ordering guarantees for paused topologies -- i.e. it's okay to still processing those tasks for a while after the pause() call is triggered. Is it really a correctness or concurrency issue?

guozhangwang · 2023-01-31T23:47:26Z

-        return tasks.allTasksPerId();
+        if (stateUpdater != null) {
+            final Map<TaskId, Task> ret = stateUpdater.getTasks().stream().collect(Collectors.toMap(Task::id, x -> x));
+            ret.putAll(tasks.allTasksPerId());


I've changed the func name slightly in another PR, so if that PR is merged we need to do a slight rebase/conflict resolution, just FYI.

Yes, once you merge the other one we can rebase this one.

guozhangwang · 2023-02-07T19:37:21Z

I do not have further comments. @cadonna do you want to take a final look? If not I can go ahead and merge it.

cadonna

@lucasbru Thanks for the updates!

I have one major comment in ReadOnlyTask.

cadonna · 2023-02-14T10:44:32Z

    @Override
    public boolean commitNeeded() {
-        throw new UnsupportedOperationException("This task is read-only");
+        return task.commitNeeded();


It gives me a bit of a headache to forward the call to commitNeeded() since it changes the state of a stream task by updating the consumed offsets.
I am also wondering why Streams also commits restoring tasks.

cadonna · 2023-02-14T10:50:36Z

    @Override
    public Map<TopicPartition, Long> changelogOffsets() {
-        throw new UnsupportedOperationException("This task is read-only");
+        return task.changelogOffsets();


If I understand the code correctly, this forwarding is fine.

guozhangwang · 2023-02-14T19:39:14Z

@lucasbru seems I forgot to send those pending comments from last week, sorry... anyways there's only one comment that I had here: #13025 (comment) and I think it's okay to no address in this PR actually, as we can do it in a different PR (I still need to update the logic a bit while I add more metrics recordings anyways). just FYI.

guozhangwang · 2023-02-17T01:11:00Z


    StreamTask activeTask(final TaskManager taskManager, final TopicPartition partition) {
-        final Stream<Task> standbys = taskManager.allTasks().values().stream().filter(Task::isActive);
+        final Stream<Task> standbys = taskManager.allOwnedTasks().values().stream().filter(Task::isActive);


For my understanding: why we need to change the callee here as well?

Good point, we don't

guozhangwang · 2023-02-17T01:13:28Z

            add("changelogPartitions");
            add("commitRequested");
            add("isActive");
-            add("commitNeeded");


Good to know :)

guozhangwang · 2023-02-17T16:37:12Z

@lucasbru the pause/ resume integration test fails again for J11/S13, could you take a look into it?

lucasbru · 2023-02-20T12:48:55Z

@lucasbru the pause/ resume integration test fails again for J11/S13, could you take a look into it?

Ah yes, that was coming from the rebase. should be fixed now

guozhangwang · 2023-02-21T18:17:14Z

LGTM! Merged to trunk.

lucasbru force-pushed the pause_resume branch from bf4b07e to 89b95be Compare December 20, 2022 16:30

cadonna reviewed Dec 20, 2022

View reviewed changes

lucasbru force-pushed the pause_resume branch from 89b95be to 3e27472 Compare January 4, 2023 15:49

lucasbru added 5 commits January 4, 2023 16:50

Minor: enable Pause/Resume integration test

50b03d2

This wasn't enabled and didn't pass for state updater.

KAFKA-14299: Make sure no progress is made on paused topologies

6649c83

The state updater restored one round of polls from the restore consumer before realizing that a newly added task was already in paused state when being added. This is tested by the existing `PauseResumeIntegrationTest`.

Bruno's comments

1c56e83

All tasks in TaskManager should return tasks in restoration

8d306e2

Wake up state updater when tasks are being resumed

827f7ec

lucasbru force-pushed the pause_resume branch from 3e27472 to 827f7ec Compare January 4, 2023 15:50

cadonna reviewed Jan 12, 2023

View reviewed changes

lucasbru changed the title ~~KAFKA-14299: Make sure no progress is made on paused topologies~~ KAFKA-14299: Fix pause and resume with state updater Jan 31, 2023

guozhangwang reviewed Jan 31, 2023

View reviewed changes

lucasbru added 2 commits February 6, 2023 11:30

Merge remote-tracking branch 'origin/trunk' into pause_resume

0846241

Reviewer comments addressed

124fedf

guozhangwang mentioned this pull request Feb 10, 2023

KAFKA-10199: Add task updater metrics, part 1 #13228

Merged

3 tasks

cadonna reviewed Feb 14, 2023

View reviewed changes

lucasbru added 3 commits February 16, 2023 13:54

Fix: only process owned tasks in maybeCommit

9d83fe7

Merge remote-tracking branch 'upstream/trunk' into pause_resume

e030650

Fix tests

7673087

guozhangwang reviewed Feb 17, 2023

View reviewed changes

lucasbru added 2 commits February 20, 2023 13:46

Fix: add state updater tasks to readOnlyAllTasks as well.

019de46

Merge remote-tracking branch 'upstream/trunk' into pause_resume

198461d

guozhangwang merged commit 0fc029c into apache:trunk Feb 21, 2023

lucasbru added the streams-thread-refactoring label Oct 23, 2025

Uh oh!

Conversation

lucasbru commented Dec 20, 2022 • edited by guozhangwang Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

lucasbru commented Dec 20, 2022

Uh oh!

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

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

Choose a reason for hiding this comment

Uh oh!

lucasbru commented Jan 31, 2023

Uh oh!

guozhangwang commented Jan 31, 2023

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!

lucasbru Feb 15, 2023 • 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!

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 7, 2023

Uh oh!

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

guozhangwang commented Feb 14, 2023

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 17, 2023

lucasbru commented Dec 20, 2022 •

edited by guozhangwang

Loading

lucasbru Feb 15, 2023 •

edited

Loading