apache / apache/texera

Workflow-completion cleanup in WorkerExecutionCompletedHandler is unreachable when evaluated, so coordinator timers keep running

Open
#7,032 2 comments 0 reactions 1 assignee Claimed by @anthonychengit View on GitHub
Dominant language
Scala
Stars
314
Forks
187
Avg merge
1d 21h
Merged PRs (30d)
214

Description

### 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).

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.