What happened?
OutputPortResultWriterThread.run() calls bufferedItemWriter.close() (which performs the iceberg commit() for the port's output table). If close() throws — for example because iceberg's built-in commit-retry budget is exhausted under heavy concurrent appends — the exception propagates out of Thread.run() and the writer thread dies silently:
override def run(): Unit = {
var internalStop = false
while (!internalStop) {
queue.take() match {
case Left(tuple) => bufferedItemWriter.putOne(tuple)
case Right(_) => internalStop = true
}
}
bufferedItemWriter.close() // ← throws → thread dies, no one notices
}OutputManager.closeOutputStorageWriterIfNeeded calls writerThread.join(), which returns normally regardless of whether the thread exited cleanly or via exception. The DP thread then fires portCompleted to the controller as if the storage was durably written.
Symptoms downstream:
- Reader-side operators (e.g., HashJoin probe reading the build-side iceberg table) see incomplete or stale data.
- The workflow may never reach
COMPLETED, or reaches it with empty results.
- The test / user observes a 1-minute
Await.result(completion, Duration.fromMinutes(1)) timeout in DataProcessingSpec.executeWorkflow with no clue why.
Expected: the writer-thread failure should propagate to the worker, then to the controller via the existing AllForOneStrategy supervisor in Controller.scala, then to the client as a FatalError — surfacing the iceberg cause immediately instead of a silent 1-minute timeout.
How to reproduce?
- Trigger any workflow whose physical plan has multiple workers writing to the same logical output port (HashJoin build/probe partitions, partitioned aggregate, shuffle output, etc.).
- Force iceberg commit to fail in
IcebergTableWriter.flushBuffer() by either:
- Running on slow CI runners where multiple workers race on the same metadata.json (observed naturally on
apache/texera/actions/runs/25259338567/job/74063902070 for csv->(csv->)->join workflow normally in DataProcessingSpec), or
- Mocking the iceberg
Table.newAppend().appendFile().commit() path to throw CommitFailedException for more retries than the configured commit.retry.num-retries (10).
- Observe: instead of a
FatalError with the iceberg cause, the test hangs for 1 minute and reports com.twitter.util.TimeoutException: 1.minutes. The workflow's regions log "successfully terminated" right before the silence begins.
Branch
main (also reproduced on release/v1.1.0-incubating)
Commit Hash (Optional)
8f14cec (release/v1.1.0-incubating backport CI for #4632)
Relevant log output
[2026-05-02 19:04:38,156] [WARN] [org.apache.iceberg.util.Tasks] [Thread-964] -
Retrying task after failure: Cannot commit operator-port-result.wid_1_eid_1_globalportid_(
logicalOpId=HashJoinOpDesc-acaa31f6-...,
layerName=build, portId=0, isInternal=true, isInput=false)_result:
metadata location ...metadata/00000-...metadata.json has changed from ...metadata/00001-...metadata.json
org.apache.iceberg.exceptions.CommitFailedException: ...
at org.apache.texera.amber.engine.architecture.worker.managers.OutputPortResultWriterThread.run(OutputPortResultWriterThread.scala:47)
...
[2026-05-02 19:04:38,743] [INFO] [CONTROLLER] [RegionExecutionCoordinator] - Region 0 successfully terminated.
... 60 seconds of silence ...
[info] - should execute csv->(csv->)->join workflow normally *** FAILED ***
[info] com.twitter.util.TimeoutException: 1.minutes
[info] at com.twitter.util.Promise.ready(Promise.scala:680)
[info] ...
[info] at org.apache.texera.amber.engine.e2e.DataProcessingSpec.executeWorkflow(DataProcessingSpec.scala:151)
Out of scope
- Adding our own retry layer on top of iceberg. Iceberg's built-in retry is already at 10 retries / 100ms→10s backoff (see
storage.conf), which is enough headroom.
- Changing the multi-worker → single-table design. Multiple workers writing the same iceberg table is by design (HashJoin build/probe partitions, etc.).
- The "Worker X is not reachable anymore, it might have crashed" warning that
removeActorRef logs for both real crashes and normal end-of-region cleanup. That noise hides real crashes today and should be cleaned up separately.
What happened?
OutputPortResultWriterThread.run()callsbufferedItemWriter.close()(which performs the icebergcommit()for the port's output table). Ifclose()throws — for example because iceberg's built-in commit-retry budget is exhausted under heavy concurrent appends — the exception propagates out ofThread.run()and the writer thread dies silently:OutputManager.closeOutputStorageWriterIfNeededcallswriterThread.join(), which returns normally regardless of whether the thread exited cleanly or via exception. The DP thread then firesportCompletedto the controller as if the storage was durably written.Symptoms downstream:
COMPLETED, or reaches it with empty results.Await.result(completion, Duration.fromMinutes(1))timeout inDataProcessingSpec.executeWorkflowwith no clue why.Expected: the writer-thread failure should propagate to the worker, then to the controller via the existing
AllForOneStrategysupervisor inController.scala, then to the client as aFatalError— surfacing the iceberg cause immediately instead of a silent 1-minute timeout.How to reproduce?
IcebergTableWriter.flushBuffer()by either:apache/texera/actions/runs/25259338567/job/74063902070forcsv->(csv->)->join workflow normallyinDataProcessingSpec), orTable.newAppend().appendFile().commit()path to throwCommitFailedExceptionfor more retries than the configuredcommit.retry.num-retries(10).FatalErrorwith the iceberg cause, the test hangs for 1 minute and reportscom.twitter.util.TimeoutException: 1.minutes. The workflow's regions log "successfully terminated" right before the silence begins.Branch
main (also reproduced on
release/v1.1.0-incubating)Commit Hash (Optional)
8f14cec (release/v1.1.0-incubating backport CI for #4632)
Relevant log output
Out of scope
storage.conf), which is enough headroom.removeActorReflogs for both real crashes and normal end-of-region cleanup. That noise hides real crashes today and should be cleaned up separately.