Skip to content

KAFKA-14299: Fix pause and resume with state updater #13025

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 12 commits into from
Feb 21, 2023
Merged

KAFKA-14299: Fix pause and resume with state updater #13025

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
50b03d2
Minor: enable Pause/Resume integration test
lucasbru Nov 21, 2022
6649c83
KAFKA-14299: Make sure no progress is made on paused topologies
lucasbru Dec 20, 2022
1c56e83
Bruno's comments
lucasbru Jan 3, 2023
8d306e2
All tasks in TaskManager should return tasks in restoration
lucasbru Jan 3, 2023
827f7ec
Wake up state updater when tasks are being resumed
lucasbru Jan 4, 2023
0846241
Merge remote-tracking branch 'origin/trunk' into pause_resume
lucasbru Feb 6, 2023
124fedf
Reviewer comments addressed
lucasbru Feb 6, 2023
9d83fe7
Fix: only process owned tasks in `maybeCommit`
lucasbru Feb 16, 2023
e030650
Merge remote-tracking branch 'upstream/trunk' into pause_resume
lucasbru Feb 16, 2023
7673087
Fix tests
lucasbru Feb 16, 2023
019de46
Fix: add state updater tasks to `readOnlyAllTasks` as well.
lucasbru Feb 20, 2023
198461d
Merge remote-tracking branch 'upstream/trunk' into pause_resume
lucasbru Feb 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1747,6 +1747,7 @@ public void resume() {
} else {
topologyMetadata.resumeTopology(UNNAMED_TOPOLOGY);
}
threads.forEach(StreamThread::signalResume);
}

/**
Expand Down
65 changes: 50 additions & 15 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,7 @@ public void run() {

private void runOnce() throws InterruptedException {
performActionsOnTasks();
resumeTasks();
restoreTasks();
checkAllUpdatingTaskStates(time.milliseconds());
waitIfAllChangelogsCompletelyRead();
Expand All @@ -140,6 +141,16 @@ private void performActionsOnTasks() {
}
}

private void resumeTasks() {
if (isTopologyResumed.compareAndSet(true, false)) {
for (final Task task : pausedTasks.values()) {
if (!topologyMetadata.isPaused(task.id().topologyName())) {
resumeTask(task);
}
}
}
}

private void restoreTasks() {
try {
changelogReader.restore(updatingTasks);
Expand Down Expand Up @@ -229,7 +240,7 @@ private void waitIfAllChangelogsCompletelyRead() throws InterruptedException {
if (isRunning.get() && changelogReader.allChangelogsCompleted()) {
tasksAndActionsLock.lock();
try {
while (tasksAndActions.isEmpty()) {
while (tasksAndActions.isEmpty() && !isTopologyResumed.get()) {
tasksAndActionsCondition.await();
}
} finally {
Expand Down Expand Up @@ -258,21 +269,39 @@ private List<TaskAndAction> getTasksAndActions() {
}

private void addTask(final Task task) {
final TaskId taskId = task.id();

Task existingTask = pausedTasks.get(taskId);
if (existingTask != null) {
throw new IllegalStateException(
(existingTask.isActive() ? "Active" : "Standby") + " task " + taskId + " already exist in paused tasks, " +
"should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. "
+ BUG_ERROR_MESSAGE);
}
existingTask = updatingTasks.get(taskId);
if (existingTask != null) {
throw new IllegalStateException(
(existingTask.isActive() ? "Active" : "Standby") + " task " + taskId + " already exist in updating tasks, " +
"should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. "
+ BUG_ERROR_MESSAGE);
}

if (isStateless(task)) {
addToRestoredTasks((StreamTask) task);
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);

@cadonna cadonna Jan 12, 2023

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.

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

@guozhangwang guozhangwang Jan 31, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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?

@lucasbru lucasbru Feb 6, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

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.

@lucasbru lucasbru Feb 6, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

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

@guozhangwang guozhangwang Feb 10, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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?

@lucasbru lucasbru Feb 20, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

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.

changelogReader.register(task.changelogPartitions(), task.stateManager());
log.debug((task.isActive() ? "Active" : "Standby")
+ " task " + taskId + " was directly added to the paused tasks.");
} else {
final Task existingTask = updatingTasks.putIfAbsent(task.id(), task);
if (existingTask != null) {
throw new IllegalStateException((existingTask.isActive() ? "Active" : "Standby") + " task " + task.id() + " already exist, " +
"should not try to add another " + (task.isActive() ? "active" : "standby") + " task with the same id. " + BUG_ERROR_MESSAGE);
}
updatingTasks.put(taskId, task);
changelogReader.register(task.changelogPartitions(), task.stateManager());
if (task.isActive()) {
log.info("Stateful active task " + task.id() + " was added to the state updater");
log.info("Stateful active task " + taskId + " was added to the state updater");
changelogReader.enforceRestoreActive();
} else {
log.info("Standby task " + task.id() + " was added to the state updater");
log.info("Standby task " + taskId + " was added to the state updater");
if (updatingTasks.size() == 1) {
changelogReader.transitToUpdateStandby();
}
Expand Down Expand Up @@ -388,12 +417,6 @@ private void checkAllUpdatingTaskStates(final long now) {
}
}

for (final Task task : pausedTasks.values()) {
if (!topologyMetadata.isPaused(task.id().topologyName())) {
resumeTask(task);
}
}

lastCommitMs = now;
}
}
Expand All @@ -411,6 +434,7 @@ private void checkAllUpdatingTaskStates(final long now) {
private final Condition restoredActiveTasksCondition = restoredActiveTasksLock.newCondition();
private final BlockingQueue<ExceptionAndTasks> exceptionsAndFailedTasks = new LinkedBlockingQueue<>();
private final BlockingQueue<Task> removedTasks = new LinkedBlockingQueue<>();
private final AtomicBoolean isTopologyResumed = new AtomicBoolean(false);

private final long commitIntervalMs;
private long lastCommitMs;
Expand Down Expand Up @@ -506,6 +530,17 @@ public void remove(final TaskId taskId) {
}
}

@Override
public void signalResume() {
tasksAndActionsLock.lock();
try {
isTopologyResumed.set(true);
tasksAndActionsCondition.signalAll();
} finally {
tasksAndActionsLock.unlock();
}
}

@Override
public Set<StreamTask> drainRestoredActiveTasks(final Duration timeout) {
final long timeoutMs = timeout.toMillis();
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ public void clearTaskTimeout() {

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

@guozhangwang guozhangwang Jan 31, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Why do we need to change these two functions?

@lucasbru lucasbru Feb 6, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

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.

@guozhangwang guozhangwang Feb 10, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Ack. That makes sense.

@cadonna cadonna Feb 14, 2023

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.

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.

@guozhangwang guozhangwang Feb 14, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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..

@lucasbru lucasbru Feb 15, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

  1. 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

}

@Override
Expand All @@ -200,7 +200,7 @@ public StateStore getStore(final String name) {

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

@cadonna cadonna Feb 14, 2023

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.

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

}

@Override
Expand Down
5 changes: 5 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,11 @@ public int hashCode() {
*/
void remove(final TaskId taskId);

/**
* Wakes up the state updater if it is currently dormant, to check if a paused task should be resumed.
*/
void signalResume();

/**
* Drains the restored active tasks from the state updater.
*
Expand Down
5 changes: 5 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1093,6 +1093,11 @@ public boolean isThreadAlive() {
return isAlive();
}

// Call method when a topology is resumed
public void signalResume() {
taskManager.signalResume();
}

/**
* Try to commit all active tasks owned by this thread.
*
Expand Down
14 changes: 13 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1115,6 +1115,12 @@ private void removeLostActiveTasksFromStateUpdater() {
}
}

public void signalResume() {
if (stateUpdater != null) {
stateUpdater.signalResume();
}
}

/**
* Compute the offset total summed across all stores in a task. Includes offset sum for any tasks we own the
* lock for, which includes assigned and unassigned tasks we locked in {@link #tryToLockAllNonEmptyTaskDirectories()}.
Expand Down Expand Up @@ -1534,7 +1540,13 @@ Set<TaskId> standbyTaskIds() {
Map<TaskId, Task> allTasks() {
// not bothering with an unmodifiable map, since the tasks themselves are mutable, but
// if any outside code modifies the map or the tasks, it would be a severe transgression.
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());

@guozhangwang guozhangwang Jan 31, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

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.

@lucasbru lucasbru Feb 6, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

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

return ret;
} else {
return tasks.allTasksPerId();
}
Comment on lines +1543 to +1549

@cadonna cadonna Jan 12, 2023

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.

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

@lucasbru lucasbru Feb 6, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Done

}

Map<TaskId, Task> notPausedTasks() {
Expand Down
2 changes: 2 additions & 0 deletions ...che/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.processor.internals.StreamThread;
import org.apache.kafka.streams.processor.internals.Task;
import org.apache.kafka.streams.processor.internals.TopologyMetadata;
import org.slf4j.Logger;
Expand Down Expand Up @@ -276,6 +277,7 @@ public boolean isNamedTopologyPaused(final String topologyName) {
*/
public void resumeNamedTopology(final String topologyName) {
topologyMetadata.resumeTopology(topologyName);
threads.forEach(StreamThread::signalResume);
}

/**
Expand Down
Loading