You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/26 10:11:52 UTC
[GitHub] [kafka] cadonna opened a new pull request, #12562: KAFKA-10199: Remove tasks from state updater on shutdown
cadonna opened a new pull request, #12562:
URL: https://github.com/apache/kafka/pull/12562
The state updater removes its updating and paused task on shutdown.
The removed tasks are added to the output queue for removed tasks.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957585278
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
Review Comment:
> That is basically just a safe guard in case somebody modifies the test setup in the future.
I see. In that case why do we not do the same in other test cases?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r955893841
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -1139,8 +1209,6 @@ public void shouldGetTasksFromUpdatingTasks() throws Exception {
stateUpdater.add(standbyTask3);
verifyUpdatingTasks(activeTask1, activeTask2, standbyTask1, standbyTask2, standbyTask3);
- final Set<Task> tasks = stateUpdater.getTasks();
-
Review Comment:
This is actually not needed since we verify the tasks in `verifyGetTasks()` below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12562:
URL: https://github.com/apache/kafka/pull/12562#issuecomment-1230134118
OK, I think I fixed the flaky test!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12562:
URL: https://github.com/apache/kafka/pull/12562#issuecomment-1230547092
Failures are unrelated
```
Build / JDK 17 and Scala 2.13 / kafka.api.TransactionsExpirationTest.testTransactionAfterProducerIdExpires(String).quorum=kraft
Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testTimeouts()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957636404
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
Review Comment:
I see. I agree that we do not need it in `shouldRemoveTasksFromAndClearInputQueueOnShutdown()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957088577
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -427,6 +430,23 @@ public void shutdown(final Duration timeout) {
stateUpdaterThread = null;
} catch (final InterruptedException ignored) {
}
+ } else {
+ removeAddedTasksFromInputQueue();
Review Comment:
Yes, this part is a bit tricky. You can enqueue tasks to add to the input queue with `StateUpdater#add()` before you start the state updater with `StateUpdater#start()`. Additionally, we have an invariant in our design that states that a task is either in the state updater or in the task registry (i.e., the `Tasks` object), but never in both at the same time.
So if you added tasks to the state updater and you shutdown the state updater before you start it, you would lose the references to those tasks. That means we would not be able to close those tasks anymore.
Tasks in the state updater can only be paused if they are already in the state updater. If the state updater has not been started, the request to add the tasks to be paused would still be in the input queue and the state updater would remove it from there. If the state updater has been started, the state updater would remove the paused tasks during shutdown.
Also note that `removeAddedTasksFromInputQueue()` is also called in the cleanup code of the state updater thread on line 108.
If you are wondering why we allow to add tasks to the state updater before we start the state updater thread, the main reason is that it simplifies testing the state updater.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957103903
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ final StreamTask statelessTask = statelessTask(TASK_0_0).inState(State.RESTORING).build();
+ final StreamTask statefulTask = statefulTask(TASK_1_0, mkSet(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ stateUpdater.pause(TASK_0_1);
+ stateUpdater.add(statelessTask);
+ stateUpdater.add(statefulTask);
+ stateUpdater.remove(TASK_1_1);
+ stateUpdater.add(standbyTask);
+ stateUpdater.resume(TASK_0_1);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(statelessTask, statefulTask, standbyTask);
+ }
+
+ @Test
+ public void shouldRemoveUpdatingTasksOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ stateUpdater = new DefaultStateUpdater(new StreamsConfig(configProps(Integer.MAX_VALUE)), changelogReader, time);
+ final StreamTask activeTask = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ when(changelogReader.completedChangelogs()).thenReturn(Collections.emptySet());
+ when(changelogReader.allChangelogsCompleted()).thenReturn(false);
+ stateUpdater.start();
+ stateUpdater.add(activeTask);
+ stateUpdater.add(standbyTask);
+ verifyUpdatingTasks(activeTask, standbyTask);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(activeTask, standbyTask);
+ verify(activeTask).maybeCheckpoint(true);
+ verify(standbyTask).maybeCheckpoint(true);
+ }
+
+ @Test
+ public void shouldRemovePausedTasksOnShutdown() throws Exception {
Review Comment:
What Guozhang said. I also explained it as an answer to one of Walker's comments above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12562:
URL: https://github.com/apache/kafka/pull/12562#issuecomment-1230034782
@guozhangwang There is a test failure for one of the added tests. So please do not merge this PR before I fixed that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957415080
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -427,6 +430,23 @@ public void shutdown(final Duration timeout) {
stateUpdaterThread = null;
} catch (final InterruptedException ignored) {
}
+ } else {
+ removeAddedTasksFromInputQueue();
Review Comment:
Alright that explains it, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r955892807
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -60,6 +60,7 @@ private class StateUpdaterThread extends Thread {
private final ChangelogReader changelogReader;
private final AtomicBoolean isRunning = new AtomicBoolean(true);
private final Map<TaskId, Task> updatingTasks = new ConcurrentHashMap<>();
+ private final Map<TaskId, Task> pausedTasks = new ConcurrentHashMap<>();
Review Comment:
I moved the map of paused tasks from the state update to the state updater thread since it actually is -- as the map of updating tasks -- an internal data structure of the thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r956444131
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -86,6 +87,10 @@ public boolean onlyStandbyTasksLeft() {
return !updatingTasks.isEmpty() && updatingTasks.values().stream().noneMatch(Task::isActive);
}
+ public Collection<Task> getPausedTasks() {
Review Comment:
Just to confirm this will not change the pause resume API?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -557,7 +577,9 @@ public Set<Task> getRemovedTasks() {
}
public Set<Task> getPausedTasks() {
- return Collections.unmodifiableSet(new HashSet<>(pausedTasks.values()));
+ return stateUpdaterThread != null
Review Comment:
We can only pause tasks with the state updated thread? What if they are using the old/different implementation? I think that can get confusing
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -427,6 +430,23 @@ public void shutdown(final Duration timeout) {
stateUpdaterThread = null;
} catch (final InterruptedException ignored) {
}
+ } else {
+ removeAddedTasksFromInputQueue();
Review Comment:
Do we also want to remove the paused tasks here? If not why do we need remove these?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ final StreamTask statelessTask = statelessTask(TASK_0_0).inState(State.RESTORING).build();
+ final StreamTask statefulTask = statefulTask(TASK_1_0, mkSet(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ stateUpdater.pause(TASK_0_1);
+ stateUpdater.add(statelessTask);
+ stateUpdater.add(statefulTask);
+ stateUpdater.remove(TASK_1_1);
+ stateUpdater.add(standbyTask);
+ stateUpdater.resume(TASK_0_1);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(statelessTask, statefulTask, standbyTask);
+ }
+
+ @Test
+ public void shouldRemoveUpdatingTasksOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ stateUpdater = new DefaultStateUpdater(new StreamsConfig(configProps(Integer.MAX_VALUE)), changelogReader, time);
+ final StreamTask activeTask = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ when(changelogReader.completedChangelogs()).thenReturn(Collections.emptySet());
+ when(changelogReader.allChangelogsCompleted()).thenReturn(false);
+ stateUpdater.start();
+ stateUpdater.add(activeTask);
+ stateUpdater.add(standbyTask);
+ verifyUpdatingTasks(activeTask, standbyTask);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(activeTask, standbyTask);
+ verify(activeTask).maybeCheckpoint(true);
+ verify(standbyTask).maybeCheckpoint(true);
+ }
+
+ @Test
+ public void shouldRemovePausedTasksOnShutdown() throws Exception {
Review Comment:
Nice test coverage! I am only confused on why we call `removeAddedTasksFromInputQueue` on shutdown and not `removeUpdatingAndPausedTasks` as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957102630
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
Review Comment:
That is basically just a safe guard in case somebody modifies the test setup in the future. For example, if somebody starts the state updater in a `setup()` method. This shutdown ensures that we have a clean and not-started state updater for this test.
In `shouldRemoveUpdatingTasksOnShutdown()` it ensures that we shutdown the state updater created on line 89 before we re-create it with a very high commit interval so that we can be sure that the checkpoint that is created was created by the clean-up code and not during normal state updating.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957073457
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -86,6 +87,10 @@ public boolean onlyStandbyTasksLeft() {
return !updatingTasks.isEmpty() && updatingTasks.values().stream().noneMatch(Task::isActive);
}
+ public Collection<Task> getPausedTasks() {
Review Comment:
No, I just moved the map of paused tasks to the state updater thread. The state updater does not have permissions to access private fields of the state udater thread due to encapsulation, so I needed to add a getter for the tasks. The behavior stays the same. You can see below in `getPausedTasks()` that the call to `pausedTasks.values()` was replaced with a call to this method which does exactly the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957076036
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -100,7 +105,8 @@ public void run() {
} catch (final RuntimeException anyOtherException) {
handleRuntimeException(anyOtherException);
} finally {
- clear();
+ removeAddedTasksFromInputQueue();
+ removeUpdatingAndPausedTasks();
Review Comment:
Yeah, this is intentional. We need to close the task that are in the the `restoredActiveTasks` queue. If I cleared that queue here, we would lose the references to those tasks and we would not be able to ever close them. I have an additional PR that moves the tasks from the output queues to the task registry where they can be closed during the shutdown of the task manager.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957088577
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -427,6 +430,23 @@ public void shutdown(final Duration timeout) {
stateUpdaterThread = null;
} catch (final InterruptedException ignored) {
}
+ } else {
+ removeAddedTasksFromInputQueue();
Review Comment:
Yes, this part is a bit tricky. You can enqueue tasks to add to the input queue with `StateUpdater#add()` before you start the state updater with `StateUpdater#start()`. Additionally, we have an invariant in our design that states that a task is either in the state updater or in the task registry (i.e., the `Tasks` object), but never in both at the same time.
So if you added tasks to the state updater and you shutdown the state updater before you start it, you would lose the references to those tasks. That means we would not be able to close those tasks anymore.
Tasks in the state updater can only be paused if they are already in the state updater. If the state updater has not been started, the request to add the tasks to be paused would still be in the input queue and the state updater would remove it from there. If the state updater has been started, the state updater would remove the paused tasks during shutdown.
Also note that `removeAddedTasksFromInputQueue()` is also called in the cleanup code of the state updater thread on line 108.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r957095777
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -557,7 +577,9 @@ public Set<Task> getRemovedTasks() {
}
public Set<Task> getPausedTasks() {
- return Collections.unmodifiableSet(new HashSet<>(pausedTasks.values()));
+ return stateUpdaterThread != null
Review Comment:
This check only verifies that the state updater thread is running. If there is no state updater running there cannot be tasks in restoration that can be paused. Additionally -- and that is the main reason for this check -- we would run into a `NullPointerException` when calling `stateUpdaterThread.getPausedTasks()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna merged pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna merged PR #12562:
URL: https://github.com/apache/kafka/pull/12562
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12562:
URL: https://github.com/apache/kafka/pull/12562#issuecomment-1228315114
Call for review: @wcarlson5 @lihaosky
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12562: KAFKA-10199: Remove tasks from state updater on shutdown
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12562:
URL: https://github.com/apache/kafka/pull/12562#discussion_r956661497
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
Review Comment:
Why we call `shutdown` first? We should have shutdown upon tearing down each test.
Ditto in `shouldRemoveUpdatingTasksOnShutdown`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -100,7 +105,8 @@ public void run() {
} catch (final RuntimeException anyOtherException) {
handleRuntimeException(anyOtherException);
} finally {
- clear();
+ removeAddedTasksFromInputQueue();
+ removeUpdatingAndPausedTasks();
Review Comment:
We no longer clear the `restoredActiveTasks`, is that intentional?
Do we assume by the time the thread is shutting down, the stream thread would not care about any restored tasks any longer, or are you going to merge the restored tasks with removed tasks soon anyways?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -125,6 +125,80 @@ public void shouldShutdownStateUpdaterAndRestart() {
verify(changelogReader, times(2)).clear();
}
+ @Test
+ public void shouldRemoveTasksFromAndClearInputQueueOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ final StreamTask statelessTask = statelessTask(TASK_0_0).inState(State.RESTORING).build();
+ final StreamTask statefulTask = statefulTask(TASK_1_0, mkSet(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ stateUpdater.pause(TASK_0_1);
+ stateUpdater.add(statelessTask);
+ stateUpdater.add(statefulTask);
+ stateUpdater.remove(TASK_1_1);
+ stateUpdater.add(standbyTask);
+ stateUpdater.resume(TASK_0_1);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(statelessTask, statefulTask, standbyTask);
+ }
+
+ @Test
+ public void shouldRemoveUpdatingTasksOnShutdown() throws Exception {
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+ stateUpdater = new DefaultStateUpdater(new StreamsConfig(configProps(Integer.MAX_VALUE)), changelogReader, time);
+ final StreamTask activeTask = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final StandbyTask standbyTask = standbyTask(TASK_0_2, mkSet(TOPIC_PARTITION_C_0)).inState(State.RUNNING).build();
+ when(changelogReader.completedChangelogs()).thenReturn(Collections.emptySet());
+ when(changelogReader.allChangelogsCompleted()).thenReturn(false);
+ stateUpdater.start();
+ stateUpdater.add(activeTask);
+ stateUpdater.add(standbyTask);
+ verifyUpdatingTasks(activeTask, standbyTask);
+ verifyRemovedTasks();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ verifyRemovedTasks(activeTask, standbyTask);
+ verify(activeTask).maybeCheckpoint(true);
+ verify(standbyTask).maybeCheckpoint(true);
+ }
+
+ @Test
+ public void shouldRemovePausedTasksOnShutdown() throws Exception {
Review Comment:
It's possible that the state updater thread was not started when shutdown was called, in that case we do not need to do the latter since it's always empty still.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org