apache / apache/texera

Multi-URI input port is declared complete while later materialization reader threads are still producing

Open
#6,919 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.