lucasbru commented on code in PR #14001:
URL: https://github.com/apache/kafka/pull/14001#discussion_r1288175911
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java:
##########
@@ -86,12 +87,29 @@ private void runOnce(final long nowMs) {
if (currentTask == null) {
currentTask =
taskManager.assignNextTask(DefaultTaskExecutor.this);
- } else {
- // if a task is no longer processable, ask task-manager to
give it another
- // task in the next iteration
- if (currentTask.isProcessable(nowMs)) {
+ }
+
+ if (currentTask != null) {
Review Comment:
I checked this out and we need to be a bit careful here to not have the task
executors to sleep when work is available. We had several bugs like this in
state updater. We have to make sure to signal on the condtion variable any time
a task becomes "processable". Here are some situations where a task becomes
processable:
- Task is unassigned from another TaskExecutor.
- Task state is changed (should only happen inside when a task is locked
inside the polling phase).
- When tasks are unlocked.
- When tasks are added.
- New records available.
- A task is resumed.
So in summary, we need to awake the task executors after every poll phase,
inside `StreamThread.signalResume`, `DefaultTaskManager.unassignTask`,
`DefaultTaskManager.unlockTasks`, `DefaultTaskManager.addTask`.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
