refactor(amber): count a materialized input port as completed once its END is processed

[aglinxinyuan]

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.
• Nothing reads the reader thread's finished flag any more, so it is removed, and it can't stand in for port completion again. The reader-thread tests that read it now check the END_CHANNEL the reader enqueues instead.

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, even before its reader thread has returned. It is not completed while END is still queued, even after the thread has returned.
• EndChannelHandlerSpec: when END arrives on a materialized input port before its reader thread has returned, 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"

InputPortMaterializationReaderThreadSpec reads real Iceberg tables. Locally only its two cases that don't need storage ran; CI runs the rest.

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

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• No candidates found from git blame history.

[github-actions]

Backport auto-label report

This fix: PR was checked against each actively-supported release branch. A release/* label nominates a backport target; the branch's release manager approving this PR is what sends the fix there. The required Backport Approvals check stays red until every label below is approved, so each manager either approves or removes their own label — which is why the labels left on a merged PR are exactly the branches it reached.

Release branch	Analysis
✅ release/v1.3	Already labeled — this fix is queued to backport here. @mengw15 decides: approving sends the fix here, removing this label declines it. The merge waits on one or the other.
✅ release/v1.2	Already labeled — this fix is queued to backport here. @xuang7 decides: approving sends the fix here, removing this label declines it. The merge waits on one or the other.

Auto-label run.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 3 worse · ⚪ 12 noise (<±5%) · 0 without baseline

Compared against main 6a29c59 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	390	0.238	25,446/33,752/33,752 us	🔴 -9.2% / 🔴 +127.5%
⚪	bs=100 sw=10 sl=64	794	0.485	122,277/157,739/157,739 us	⚪ within ±5% / 🔴 +56.7%
⚪	bs=1000 sw=10 sl=64	918	0.56	1,095,180/1,160,050/1,160,050 us	⚪ within ±5% / 🔴 +21.2%

Baseline details

Latest main 6a29c59 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	390 tuples/sec	429 tuples/sec	840.06 tuples/sec	-9.1%	-53.6%
bs=10 sw=10 sl=64	MB/s	0.238 MB/s	0.262 MB/s	0.513 MB/s	-9.2%	-53.6%
bs=10 sw=10 sl=64	p50	25,446 us	23,351 us	11,813 us	+9.0%	+115.4%
bs=10 sw=10 sl=64	p95	33,752 us	32,381 us	14,838 us	+4.2%	+127.5%
bs=10 sw=10 sl=64	p99	33,752 us	32,381 us	17,837 us	+4.2%	+89.2%
bs=100 sw=10 sl=64	throughput	794 tuples/sec	821 tuples/sec	1,087 tuples/sec	-3.3%	-26.9%
bs=100 sw=10 sl=64	MB/s	0.485 MB/s	0.501 MB/s	0.663 MB/s	-3.2%	-26.9%
bs=100 sw=10 sl=64	p50	122,277 us	117,973 us	94,074 us	+3.6%	+30.0%
bs=100 sw=10 sl=64	p95	157,739 us	152,228 us	100,688 us	+3.6%	+56.7%
bs=100 sw=10 sl=64	p99	157,739 us	152,228 us	116,015 us	+3.6%	+36.0%
bs=1000 sw=10 sl=64	throughput	918 tuples/sec	923 tuples/sec	1,128 tuples/sec	-0.5%	-18.6%
bs=1000 sw=10 sl=64	MB/s	0.56 MB/s	0.563 MB/s	0.689 MB/s	-0.5%	-18.7%
bs=1000 sw=10 sl=64	p50	1,095,180 us	1,086,170 us	916,643 us	+0.8%	+19.5%
bs=1000 sw=10 sl=64	p95	1,160,050 us	1,208,131 us	957,169 us	-4.0%	+21.2%
bs=1000 sw=10 sl=64	p99	1,160,050 us	1,208,131 us	978,578 us	-4.0%	+18.5%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,512.88,200,128000,390,0.238,25445.79,33752.08,33752.08
1,100,10,64,20,2518.90,2000,1280000,794,0.485,122276.92,157738.96,157738.96
2,1000,10,64,20,21796.76,20000,12800000,918,0.560,1095179.64,1160050.12,1160050.12

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.61%. Comparing base (6a29c59) to head (bb46d20).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #8938      +/-   ##
============================================
- Coverage     92.61%   92.61%   -0.01%     
+ Complexity     5059     5056       -3     
============================================
  Files          1258     1258              
  Lines         53780    53770      -10     
  Branches       6707     6705       -2     
============================================
- Hits          49811    49798      -13     
- Misses         2313     2316       +3     
  Partials       1656     1656              

Flag	Coverage Δ		*Carryforward flag
access-control-service	77.38% <ø> (ø)		Carriedforward from 9cf43f2
agent-service	99.16% <ø> (ø)		Carriedforward from 9cf43f2
amber	88.10% <100.00%> (-0.01%)	⬇️
computing-unit-managing-service	60.48% <ø> (ø)		Carriedforward from 9cf43f2
config-service	87.37% <ø> (ø)		Carriedforward from 9cf43f2
file-service	82.00% <ø> (ø)		Carriedforward from 9cf43f2
frontend	96.59% <ø> (+<0.01%)	⬆️	Carriedforward from 9cf43f2
notebook-migration-service	83.73% <ø> (ø)		Carriedforward from 9cf43f2
pyamber	98.52% <ø> (-0.07%)	⬇️	Carriedforward from 9cf43f2
workflow-compiling-service	72.43% <ø> (ø)		Carriedforward from 9cf43f2

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[carloea2]

Retested bb46d20 locally on Windows with JDK 17, Python 3.12 and isolated PostgreSQL databases. All 118 focused engine tests and all 8 LoopIntegrationSpec tests pass. Formatting also passes.

sbt -batch 'WorkflowExecutionService/testOnly *InputManagerSpec *EndChannelHandlerSpec *StartHandlerSpec *DataProcessorSpec *EndHandlerSpec *StartChannelHandlerSpec *AssignPortHandlerSpec *DPThreadSpec *WorkerSpec *InputPortMaterializationReaderThreadSpec -- -oDF' 'scalafmtCheckAll' 'WorkflowExecutionService/testOnly *LoopIntegrationSpec -- -oDF'

This fixes the completion race I reproduced. No competing PR is needed. Backport approvals are still pending.

[carloea2]

Verified bb46d20 locally. All 118 focused tests and 8 loop tests pass, formatting is clean, and required build checks are green. The completion race is fixed. Backport approvals from the release managers are still pending.

[carloea2]

@Xiao-zhen-Liu, required checks now pass on Linux and macOS, and the review thread is resolved. Could you merge this when you have time? I can then refresh #8142 and #8144 to trigger CI with the loop fix.