aglinxinyuan opened a new issue, #6919: URL: https://github.com/apache/texera/issues/6919
### What happened? A multi-URI input port can be declared complete while all but the first of its materialization reader threads are still producing data. `InputManager.isPortCompleted` consults only the **head** of the reader-thread list ([InputManager.scala:121-131](https://github.com/apache/texera/blob/main/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala#L121-L131)): ```scala val existingThread = this.inputPortMaterializationReaderThreads(portId).head existingThread.finished ``` but `setupInputPortMaterializationReaderThreads` deliberately builds a **list** — one thread per materialization URI (`InputManager.scala:77-88`). When a port reads from ≥2 URIs and the first reader finishes early: | step | consequence | |---|---| | `.head.finished` becomes true | port reported complete while reader #2 still calls `inputMessageQueue.put(...)` | | `EndChannelHandler` finalizes output → `portCompleted` → region completion | coordinator sends `EndWorker` while real `DataFrame`s are still being enqueued | | worker is `gracefulStop`ped | remaining tuples from URI #2 are dropped | Two aggravating details: 1. The completion flag flips **after** the last enqueue but before the DP thread has consumed it: `emitECM(METHOD_END_CHANNEL, ...); isFinished.set(true)` (`InputPortMaterializationReaderThread.scala:143-144`). 2. Nothing ever interrupts or joins these threads — `WorkflowWorker.postStop` / `DPThread.stop` don't touch them, and they inherit non-daemon status from the DP thread executor, so a still-running reader outlives the actor and keeps writing into a queue nobody drains (also keeps test JVMs alive). Multi-URI ports are reachable by construction: `ExpansionGreedyScheduleGenerator` groups `matReaderWriterPairs` by input port and maps **all** paired upstream output ports to URIs, and `ResourceAllocator` zips them into `InputPortConfig(uris.zip(portPartitionings))`. ### How to reproduce? Run a workflow where one input port is fed by two materialized upstream output ports (two URIs) of very different sizes; complete the small one first. `isPortCompleted` returns true as soon as reader #1 finishes, regardless of reader #2's progress. ### Version/Branch main (observed at 429be110a7; discovered during the investigation for #6916). -- 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]
