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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
