Multi-URI input port is declared complete while later materialization reader threads are still producing
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### 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).
Contributor guide
Research direction
Start with InputManager.scala:77-88 and 121-131, then read InputPortMaterializationReaderThread.scala:143-144 to trace reader completion and queueing. Inspect EndChannelHandler, WorkflowWorker.postStop, and DPThread.stop for downstream completion and shutdown behavior. Reproduce with a multi-URI input whose readers finish at different times; done means the port and worker do not complete or discard data while any reader remains active.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100