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 1 commit
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
2 changes: 1 addition & 1 deletion 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() {
return task.commitNeeded();
throw new UnsupportedOperationException("This task is read-only");
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion 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 @@ -1115,7 +1115,7 @@ int maybeCommit() {
}

committed = taskManager.commit(
taskManager.allTasks()
taskManager.allOwnedTasks()
.values()
.stream()
.filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
Expand Down
11 changes: 11 additions & 0 deletions 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 @@ -1549,6 +1549,17 @@ Map<TaskId, Task> allTasks() {
}
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

}

/**
* Returns tasks owned by the stream thread. With state updater disabled, these are all tasks. With
* state updater enabled, this does not return any tasks currently owned by the state updater.
* @return
*/
Map<TaskId, Task> allOwnedTasks() {
// 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();
}

Map<TaskId, Task> notPausedTasks() {
return Collections.unmodifiableMap(tasks.allTasks()
.stream()
Expand Down
1 change: 0 additions & 1 deletion streams/src/test/java/org/apache/kafka/streams/processor/internals/ReadOnlyTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ class ReadOnlyTaskTest {
add("changelogPartitions");
add("commitRequested");
add("isActive");
add("commitNeeded");

@guozhangwang guozhangwang Feb 17, 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.

Good to know :)

add("changelogOffsets");
add("state");
add("id");
Expand Down
16 changes: 16 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -513,6 +513,22 @@ public void shouldReturnStateUpdaterTasksInAllTasks() {
assertEquals(taskManager.allTasks(), mkMap(mkEntry(taskId03, activeTask), mkEntry(taskId02, standbyTask)));
}

@Test
public void shouldNotReturnStateUpdaterTasksInOwnedTasks() {
final StreamTask activeTask = statefulTask(taskId03, taskId03ChangelogPartitions)
.inState(State.RUNNING)
.withInputPartitions(taskId03Partitions).build();
final StandbyTask standbyTask = standbyTask(taskId02, taskId02ChangelogPartitions)
.inState(State.RUNNING)
.withInputPartitions(taskId02Partitions).build();
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);

when(stateUpdater.getTasks()).thenReturn(mkSet(standbyTask));
when(tasks.allTasksPerId()).thenReturn(mkMap(mkEntry(taskId03, activeTask)));
assertEquals(taskManager.allOwnedTasks(), mkMap(mkEntry(taskId03, activeTask)));
}

@Test
public void shouldCreateActiveTaskDuringAssignment() {
final StreamTask activeTaskToBeCreated = statefulTask(taskId03, taskId03ChangelogPartitions)
Expand Down