aglinxinyuan opened a new pull request, #8938: URL: https://github.com/apache/texera/pull/8938
### What changes were proposed in this PR? `InputManager.isPortCompleted` now treats a port read from materialization the same as a port fed by a link: it counts as completed once END_CHANNEL has been processed on it. `EndChannelHandler` marks that by setting `WorkerPort.completed`. Before, such a port reported its reader thread's `finished` flag. The thread sets that flag only after enqueuing END. So a DP thread waiting on the queue, typically the worker that got no tuples, could run `EndChannelHandler` before the flag was set. It then saw the port as not completed and skipped `finalizeOutput`. Nothing called it again, so the worker never finished and its region hung until the test deadline. That is the intermittent `TimeoutException: 1.minutes+30.seconds` in `LoopIntegrationSpec`. ``` Before: reader enqueues END -> DP processes END, flag not set yet -> output never finalized -> region hangs After: reader enqueues END -> DP processes END, port completed -> output finalized -> region completes ``` Two related effects of the change: - The flag could also be set while END was still queued. A worker with a second materialized input no longer finalizes before that port's END has been processed. - `startInputPortReaderThreads` uses the same check to skip ports that already completed. The second `StartWorker` of a two-phase region arrives only after the coordinator has the port's `PortCompleted`, which the worker sends after processing END. So it skips the same ports as before. ### Any related issues, documentation, discussions? Closes #8937. The two `LoopIntegrationSpec` cases added in #8639 hit the race most often; the race itself dates from #3460. ### How was this PR tested? Tests first: the first commit adds the tests and fails on `main`, and the second commit makes them pass. - `InputManagerSpec`: a materialized port is completed once END has been processed, before the reader thread sets its flag. It is not completed while END is still queued, even after the flag is set. - `EndChannelHandlerSpec`: when END arrives on a materialized input port before its reader thread sets the flag, the worker emits the output `FinalizePort` and `FinalizeExecutor`. On `main` it emits only the input port's `FinalizePort`, which is the hang. ``` sbt "WorkflowExecutionService/testOnly *InputManagerSpec *EndChannelHandlerSpec *StartHandlerSpec *DataProcessorSpec *EndHandlerSpec *StartChannelHandlerSpec *AssignPortHandlerSpec *DPThreadSpec *WorkerSpec" sbt "WorkflowExecutionService/scalafmtCheckAll" "WorkflowExecutionService/scalafixAll --check" ``` `LoopIntegrationSpec`, run locally on Windows with the sbt JVM and its Python workers pinned to 3 CPUs (affinity mask `0x7`), one case per JVM: | Case | `main` | This PR | |---|---|---| | bind a `$i` reference after two branches of the loop body meet again | 3 of 5 runs hung on the first attempt; 1 also hung on the retry and failed as on CI | 10 of 10 passed on the first attempt, 42–44 s each | | bind the inner loop's variable where it shadows the outer loop's | (fails on CI, see #8937) | 10 of 10 passed on the first attempt, 64–70 s each | Unpinned, the whole spec passes with this PR: 8 of 8 cases, 29–59 s each, no retries. ``` AMBER_TEST_FILTER= STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec -- -z \"after two branches of the loop body meet again\"" AMBER_TEST_FILTER=integration-only STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec" ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
