aglinxinyuan opened a new issue, #7032:
URL: https://github.com/apache/texera/issues/7032
### What happened?
The workflow-completion cleanup in `WorkerExecutionCompletedHandler` is
unreliable: its precondition is normally false at the moment it is evaluated,
so the coordinator's periodic timers usually keep running after a workflow has
finished.
```scala
// WorkerExecutionCompletedHandler.scala:60-74 — the Future is in statement
position (discarded)
Future.collect(Seq(statsRequest)).flatMap(_ => {
val isWorkflowTerminal =
cp.workflowExecution.isCompleted &&
!cp.workflowScheduler.hasPendingRegions &&
!cp.workflowExecutionManager.hasUnfinishedRegionManagers
if (isWorkflowTerminal) {
sendToClient(ExecutionStateUpdate(cp.workflowExecution.getState))
cp.coordinatorTimerService.disableStatusUpdate()
cp.coordinatorTimerService.disableRuntimeStatisticsCollection()
}
})
```
`hasUnfinishedRegionManagers` is
`regionExecutionManagers.values.exists(!_.isCompleted)`
(`WorkflowExecutionManager.scala:200-202`), and a `RegionExecutionManager` only
reaches `Completed` **after** `EndWorker` plus `gracefulStop` have finished for
all of its workers (`setPhase(Completed)` runs in the continuation after
`terminateWorkersWithRetry`). A worker emits `workerExecutionCompleted`
*before* it is sent `EndWorker`, so when this block runs — one statistics round
trip later — its own region is still terminating and the condition is false.
Two consequences:
1. **The timers are usually never disabled.** A repo-wide search shows only
two callers of `disableStatusUpdate`/`disableRuntimeStatisticsCollection`: this
block, and `PauseHandler` (for pause). Nothing stops them on the completion
path, so periodic status updates and statistics collection continue for the
life of the coordinator actor. Impact is modest — completed operators are
skipped during collection (`QueryWorkerStatisticsHandler.scala:150-153`), so
the traversal mostly yields empty queries — but it still emits needless client
updates and can log `unknown identifier` warnings for removed workers.
2. **It can also never run at all.** The block hangs off
`Future.collect(Seq(statsRequest))`, and that statistics chain has no timeout;
if any queried worker has already been stopped, the chain never resolves.
Workflow completion itself is still reported —
`WorkflowExecutionManager.scala:150` sends `ExecutionStateUpdate` under a
`completionNotified` CAS — so the user-visible completion event does not depend
on this block. Only the timer cleanup (and a duplicate state update) do.
Worth deciding whether the cleanup belongs here at all, or in the
region/workflow termination path where "all regions are Completed" is actually
known.
### How to reproduce?
Code inspection is sufficient for the ordering argument. Observable check:
run any workflow to completion and confirm the coordinator keeps emitting
periodic statistics/status activity afterwards (the `disable*` calls never
being reached can be confirmed with a breakpoint or a temporary log line in the
`isWorkflowTerminal` branch).
### Version/Branch
main (observed at a61702fd10; noticed while reviewing #6960).
--
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]
