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 - we should probably lock tasks when they are paused and unlock them when they are resumed. This belongs to the StreamThread integration work. - need to awake the task executors `DefaultTaskManager.unassignTask`, `DefaultTaskManager.unlockTasks`, `DefaultTaskManager.addTask`. - awake the task executors after every polling phase. -- 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