[nzw921rx]

Purpose of this pull request

This contribution to Apache SeaTunnel reduces MultiTableSink timer flush
latency and fixes a pre-existing deadlock in checkpoint preparation.

Run timer flushes in parallel across writer queues

In a 45-second JDBC sink profile with five writer queues and a 200 ms
flush interval, the dispatcher spent 39.58 seconds waiting for INSERT
responses inside aggregatedFlush. Draining pending writes took
0.30 seconds.

After draining pending writes, this change uses the existing executor to
flush independent queues concurrently. Child writers sharing a queue
flush sequentially under that queue's lock. A successful flush round
completes before the dispatcher processes more rows.

Fix a prepareCommit deadlock during table isolation

Under CONTINUE_OTHER_TABLES, a prepare worker previously called
handleTableFailure() while holding its queue lock. The handler acquired
the writer monitor and then the queue locks to remove the failed table's
writers. Two concurrent failures could form a cycle and stall the
checkpoint.

Workers now record ordinary table failures after retries are exhausted.
Once the queue tasks finish and release their locks, the checkpoint
thread isolates the recorded tables before returning commit information.
Fatal row-error handling failures still propagate after the worker
releases its queue lock.

Related to #10801, #11537, and #10600.

Does this PR introduce any user-facing change?

Yes. Multi-table jobs with independent writer queues can complete timer
flush rounds sooner. Under CONTINUE_OTHER_TABLES, healthy tables can
finish checkpoint preparation when failures in multiple queues require
table isolation.

How was this patch tested?

• Compared serial and parallel timer flush using the JMH benchmark in
  [Test][Benchmark] Add MultiTableSink timer flush benchmark #13210. With five writer queues and a 100 ms trigger interval, the
  flush callback went from 51.2 ms to 10.8 ms per round.
• Added a regression test that synchronizes two prepare failures on
  separate queues. It timed out on the old code. With the fix, both
  failed tables were isolated and the healthy table returned commit
  information.
• All 46 multi-table API tests passed on JDK 17, including parallel
  flush, failure isolation, retry, and fatal error handling.
• API module verify, Spotless, and git diff --check passed.

[DanielLeens]

Thanks a lot for this, @nzw921rx. The profiling numbers in the description (about 40 s of a 45 s window spent waiting on serial INSERTs inside aggregatedFlush) make the motivation very clear, and the close/isolation coordination shows you thought carefully about the shutdown races. I reviewed the current head 20eb548f5bf5538caa2924b86acb23148987fbbb in full against dev (535d4237dd4), tracing the real call chain from the timer through the sink task into the child writers.

What Problem Does This PR Solve?

A Multi-Table sink registers one aggregated timer flush action. On dev, MultiTableSinkWriter.aggregatedFlush flushes every child writer one after another on the sink task thread, so with N writer queues the dispatcher waits for N sequential round trips per tick. This PR drains the queues as before, then flushes the queues in parallel on the existing writer executor (still serial within a queue under that queue's lock), waits for the whole round on success, and stops the round early on a fatal error, a cancellation or close().

1 Code Change Review

1.1 Core logic

Normal path check: yes, the normal user path hits this change. Any multi-table job with a child sink that registers a flush action (JDBC, StarRocks, Elasticsearch, MongoDB, Hudi, ...) and a flush_interval/timer flush configured goes through aggregatedFlush on every tick.

TaskExecutionService.registerTimerFlushTask (TaskExecutionService.java:1098)
  -> SourceFlowLifeCycle.onTimerTick (SourceFlowLifeCycle.java:213, registered at :403-415) offers FlushSignal into the data path
  -> SinkFlowLifeCycle.processSignal (SinkFlowLifeCycle.java:777-790), sink TASK thread
       writerContext.getFlushAction().run()            // ErrorHandlingSinkWriter wrapper
         -> MultiTableSink lambda (MultiTableSink.java:272)
           -> MultiTableSinkWriter.aggregatedFlush (MultiTableSinkWriter.java:979)
                checkQueueRemain(); subSinkErrorCheck();   // queues drained, consumers idle
                activeTimerFlush CAS null -> completion   (:984)
                queueSize == 1 : flushQueue(0) on task thread (:991)
                queueSize  > 1 : submitQueueFlushes -> executorService (:1025)
                    worker q: for id in queue q:
                        synchronized(runnable[q]) { if done -> return; skip failed; child.flush() } (:1056)
                        on Exception + CONTINUE_OTHER_TABLES:
                            synchronized(this) { if done -> return; handleTableFailure } (:1078)
                    last queue -> completion.complete(null); any Throwable -> complete(error)
                completion.get()   // task thread blocks; no write/snapshot/prepareCommit can interleave
                failure/interrupt -> completion.cancel; tasks.cancel(true) (:1018-1019)
  -> drainCollectedTerminalWriteOutcomes() after the round returned (happens-before via the future)
close() (:872): activeTimerFlush <- CLOSED and complete(round) -> shutdownNow
        -> [CONTINUE] synchronized(this){} barrier waits for in-flight isolation (:888)
        -> per-queue lock + close children (waits for that queue's in-flight flush)

Key points I verified:

• Ordering with checkpoints is preserved. The round runs inside processSignal on the task thread and the success path blocks on completion.get() until every queue returns, so snapshotState/prepareCommit(long) (also task-thread calls) can never overlap a successful round, and the deferred terminal outcomes drained afterwards are visible.
• Lock order is safe for the new path. Workers release their queue lock before entering synchronized(this) for handleTableFailure, which then takes every queue lock inside removeTableWriters (:1297-1302). No thread holds a queue lock while waiting for this, so there is no cycle. The consumer threads also call the failure handler after releasing their own monitor (MultiTableWriterRunnable.java:146-149, :341-350).
• The close barrier is correct: close() completes the round before acquiring this, and a worker re-checks completion.isDone() under the same monitor before starting isolation, so either the worker sees the stop signal or close() waits for the isolation to finish. This prevents a child writer being closed twice.
• Thread budget: the pool is still queueSize * 2; queueSize long-lived consumers plus at most queueSize flush tasks per round, and flush and prepareCommit are never submitted concurrently on the success path. Since child writers are already driven from pool threads by the consumers and by prepareCommit, moving the child flush onto a pool thread does not introduce a new thread-affinity assumption, and cross-writer concurrency during a flush is the same as during normal parallel writes.

1.2 Compatibility

• No public API, config option, default value, or serialized state changes. aggregatedFlush is package-private and MultiTableState/MultiTableCommitInfo are untouched, so checkpoint/savepoint restore is unaffected.
• Behavioral change (documented in en/zh): under FAIL_FAST, other queues may already have flushed, or be mid-flush, when the first failure is reported; on dev later queues were never flushed. For at-least-once sinks this does not change the recovery contract (restore replays from the last checkpoint either way), and two-phase/XA sinks do not commit on timer flush. Which error is reported first is now nondeterministic when several queues fail at once.
• A flush after close() now fails fast with IOException("... already running or the writer is closing") instead of calling flush on closed child writers, which is an improvement.

1.3 Side effects

• Per tick on the multi-queue path: one CompletableFuture, one ArrayList, and queueSize task submissions. At typical intervals (hundreds of ms) this is negligible next to the network round trip it removes.
• Peak downstream concurrency during a flush rises from 1 to queueSize, but that equals the concurrency the consumer threads already produce during normal writes, and MultiTableResourceManager pools are already sized by queueSize, so no extra connection demand is created.
• On a failed or cancelled round, aggregatedFlush returns while other queues' flushes may still be running (they are interrupted via cancel(true)); close() then waits for each queue's in-flight flush through the queue lock. This is the intended behavior and is covered by a test.

1.4 Error handling / logging

Exceptions are rethrown with their original instance (asserted with assertSame in the new test), Error is preserved, and the interrupt flag is restored on the dispatcher. Two small observability gaps are listed below (Issue 2).

Issue 1: pre-existing lock-order deadlock in parallel prepareCommit under CONTINUE_OTHER_TABLES (not introduced by this PR)

• Location: seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java:760 and :785, together with :1134 and :1297-1302
• Problem: each prepareCommit task holds runnable[i] and then calls handleTableFailure (monitor this), which acquires every other queue lock in removeTableWriters. If two queues fail prepareCommit at the same time, task 0 holds R0 + this and waits for R1, while task 1 holds R1 and waits for this.
• Risk: checkpoint hangs forever in continue-other-tables mode when two tables on different queues fail prepareCommit concurrently. This is exactly the pattern your flush path correctly avoids by releasing the queue lock before isolation.
• Best improvement: out of scope here; a separate follow-up that collects the failure under the queue lock and isolates after releasing it (as flushQueue now does) would fix it. Happy to open an issue if you prefer.
• Severity: Medium (pre-existing, non-blocking for this PR)
• Raised by another reviewer: No

Issue 2: child flush errors after the round has ended are dropped without a log line

• Location: MultiTableSinkWriter.java:1041-1043 (completion.complete(error) returns false for every error after the first) and :1078-1081 (if (completion.isDone()) return; skips isolation of the failed table)
• Problem: when several queues fail in the same round, only the first error is propagated; the later ones are silently discarded. In continue-other-tables mode, a table whose flush failed after the round was stopped is neither isolated nor logged.
• Risk: the job still fails because of the first error, so this is not a correctness issue, but operators lose the second table's root cause when diagnosing a multi-table failure.
• Best improvement: log the dropped error at WARN with the table id (or addSuppressed it onto the first failure) when complete(error) returns false or when isolation is skipped because the round is done.
• Severity: Low
• Raised by another reviewer: No

Issue 3: cancel(true) interrupts healthy queues' in-flight flushes on FAIL_FAST

• Location: MultiTableSinkWriter.java:1019
• Problem: on the first failure, other queues' child flushes that are mid I/O get interrupted. Some clients react to interrupts by closing channels (e.g. NIO-based clients, Hadoop ClosedByInterruptException).
• Risk: low, since the task is failing anyway and recovery restarts from the last checkpoint; it can only add noisier secondary errors in logs. On dev those queues would not have been touched at all.
• Best improvement: either keep it (reasonable for prompt failure) and mention it in one sentence in the PR description, or use cancel(false) and rely on the completion.isDone() check before each child to stop the round; close() already interrupts through shutdownNow() when the task is torn down.
• Severity: Low
• Raised by another reviewer: No

2 Quality

2.1 Code standards and comments

Comments explain why (lock order, close barrier, rejected submit race), which is what this code needs. Two small nits:

Issue 4: outdated Javadoc on the registration side and on close()

• Location: seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSink.java:259-263; MultiTableSinkWriter.java:856-872
• Problem: registerAggregatedFlushIfNeeded still says the action "calls each sub-writer's flush action under the corresponding lock, ensuring safe execution from the engine timer thread"; it now runs queues in parallel, and it runs on the sink task thread via FlushSignal, not on the timer thread. The close() Javadoc does not mention that it stops an active flush round and waits for in-flight isolation.
• Best improvement: update both Javadocs in one or two sentences.
• Severity: Low

The empty synchronized (this) {} at MultiTableSinkWriter.java:888 is correct, but its correctness depends on every isolation entry re-checking completion.isDone() under the same monitor. A tiny helper such as awaitInFlightTimerFlushIsolation() would make the intent harder to break in future edits. Optional.

2.2 Tests

• testAggregatedFlushIsolatesConcurrentFailuresAndFlushesHealthyTable (MultiTableSinkWriterTest.java:577): the two failing children wait on a shared latch, so a serial implementation fails the latch assertion; it also proves concurrent isolation does not deadlock. Stability: Stable (latches plus a 10 s preemptive timeout).
• testAggregatedFlushReportsFailureAndClosesQueuesIndependently (:633): proves the original exception instance propagates while another queue is still flushing, and that close() closes queue 0 while waiting for queue 1's in-flight flush. Stability: Stable (CompletableFuture.join is not interruptible, so the shutdownNow() interrupt cannot make it flaky).
• testInterruptedAggregatedFlushDoesNotStartAnotherChildFlush (:722): both identifiers (index 0 and 2) are on queue 0, so it covers the same-queue stop-after-interrupt path and that the interrupt flag is restored. Stability: Stable.
• The old testAggregatedFlushFailFastStopsAtFirstFailure asserted that the healthy table was never flushed; that assertion no longer matches the intended semantics, so replacing it is justified rather than a weakening. One gap remains: there is no test that a FAIL_FAST failure on one queue prevents a later child on a different queue from starting (the new cross-queue completion.isDone() check). A small case, with queue 0 failing and queue 1 holding two children where the second must not start, would lock this in. Non-blocking.
• The description notes that post-change OceanBase throughput is still pending. If you can share the before/after number for the JDBC multi-table job once you have it, that would be great evidence for the performance claim. Non-blocking.

2.3 Docs

docs/en/engines/zeta/timer-flush.md and docs/zh/engines/zeta/timer-flush.md are both updated and say the same thing. The existing note that timer-flush-pool-size does not parallelize connector flushes is still accurate, because parallelism now comes from the writer queues, not from the timer pool.

3 Architecture

The change stays inside MultiTableSinkWriter, reuses the executor and queue locks that prepareCommit already relies on, and keeps per-queue (and therefore per-writer) ordering. No connector-specific logic leaks into the API layer. This is the right place for the fix.

4 Issue Summary

#	Issue	Location	Severity	Blocking
1	Pre-existing prepareCommit lock-order deadlock in continue-other-tables mode	MultiTableSinkWriter.java:760,785,1134,1297	Medium	No (follow-up)
2	Later child flush errors are dropped without logging	MultiTableSinkWriter.java:1042,1078	Low	No
3	cancel(true) interrupts healthy in-flight flushes on FAIL_FAST	MultiTableSinkWriter.java:1019	Low	No
4	Outdated Javadoc on registerAggregatedFlushIfNeeded / close()	MultiTableSink.java:259-263, MultiTableSinkWriter.java:856-872	Low	No
-	Optional: cross-queue FAIL_FAST stop test	MultiTableSinkWriterTest.java	Low	No

5 Merge Recommendation

Ready to merge after fixes

• Non-CI blockers: none. Issues 1-4 are non-blocking suggestions.
• Remaining blocker: CI has not actually run on this head. The fork Build run for 20eb548f ended immediately with zero jobs and GitHub reports it as a workflow startup failure (https://github.com/nzw921rx/seatunnel/actions/runs/42672870253); the runs for earlier heads were all cancelled by newer pushes. Could you please re-run that workflow (or push an empty commit) so the full suite runs on the current head? The branch is up to date with dev and has no conflicts. I will be glad to approve once CI is green.

Thanks again for the careful work on the shutdown and isolation edge cases.

[DanielLeens]

Thanks for the quick follow-up, @nzw921rx. I re-reviewed the current head d209c3510210168e62a1d8d42a270bb5f5c4c366 in full. Since my last review (20eb548f) the substantive change is e549c246 ("Report late multi-table timer flush failures"), plus an empty retrigger commit. I went through the new flushQueue error handling branch by branch, re-traced the close/isolation coordination, and re-checked every point from my previous review.

What Problem Does This PR Solve?

On dev, MultiTableSinkWriter.aggregatedFlush flushes every child writer serially on the sink task thread, so a multi-table sink with N writer queues waits for N sequential round trips per timer tick. This PR drains the queues as before, then flushes independent queues in parallel on the existing writer executor (still serial within a queue under that queue's lock), waits for the whole round on success, and stops the round early on a fatal error, a cancellation or close().

1 Code Change Review

1.1 Core logic

Normal path check: yes. Every multi-table job whose child sink registers a flush action (JDBC, StarRocks, Elasticsearch, ...) and has timer flush enabled goes through aggregatedFlush on every tick. The runtime flow is unchanged from my previous review:

SourceFlowLifeCycle.onTimerTick -> FlushSignal -> SinkFlowLifeCycle.processSignal (sink TASK thread)
  -> MultiTableSink aggregated flush action -> MultiTableSinkWriter.aggregatedFlush (:978)
       checkQueueRemain(); subSinkErrorCheck();       // queues drained
       activeTimerFlush CAS null -> completion       (:983)
       1 queue : flushQueue(0) on task thread
       N queues: submitQueueFlushes -> executorService (:1024)
           worker q: for id in queue q:
               synchronized(runnable[q]) { if done -> return; skip failed; child.flush() }
               catch Throwable (:1073):
                 InterruptedException          -> restore flag, complete(error) or WARN
                 Exception + CONTINUE_OTHER    -> synchronized(this){ if !done: isolate; continue }
                                                  else fall through
                 otherwise / round already done -> complete(error); if it lost the race -> WARN (:1090-1095)
           last queue -> complete(null); wrapper Throwable -> complete(error) or WARN (:1040-1046)
       completion.get(); failure -> completion.cancel(false); tasks.cancel(true) (:1017-1018)
close(): CLOSED + complete(round) -> shutdownNow -> [CONTINUE] synchronized(this){} barrier -> per-queue close

What I verified for e549c246:

• Semantics are preserved for every error class. FAIL_FAST exceptions still complete the round with the original instance (single-queue path rethrows it via completion.get(), still covered by assertSame). An Error in continue mode is still not isolated and still fails the round. InterruptedException in the single-queue path now reaches completion.complete(error) instead of being thrown, but CompletableFuture.get() on an already-completed future returns without checking the interrupt flag, so aggregatedFlush rethrows it through the catch (InterruptedException) branch with the flag restored, exactly as before.
• An exception thrown by handleTableFailure itself (the fatal row-error-handling case) is raised inside the catch block, so it is not re-caught by the new catch (Throwable); it still escapes flushQueue and fails the round, as before.
• Interrupts can no longer cause a healthy table to be isolated by mistake: both the FAIL_FAST path (completion.cancel(false) before task.cancel(true), :1017-1018) and close() (complete(...) at :874 before shutdownNow()) mark the round done before any worker is interrupted, and isolation re-checks completion.isDone() under this.
• The close barrier still holds: isolation is only entered when !completion.isDone() under the same monitor that close() acquires after signalling the round, so a child writer cannot be closed twice.

1.2 Compatibility

No public API, config option, default value or serialized state changes; MultiTableState/MultiTableCommitInfo are untouched, so checkpoint/savepoint restore is unaffected. The documented behavior change (under FAIL_FAST other queues may already have flushed when the first error is reported) is unchanged and described in en/zh docs.

1.3 Side effects

Unchanged from the previous review: one CompletableFuture, one list and queueSize submissions per tick on the multi-queue path, and peak downstream concurrency during a flush equal to what the consumer threads already produce during normal writes. The new WARN lines only fire when a round has already failed or is being closed, so they cannot add log volume on the healthy path.

1.4 Error handling / logging

Issue A (new, small): expected interrupts are now logged at WARN with a full stack trace

• Location: seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java:1074-1075 together with :1089-1095
• Problem: when a FAIL_FAST round fails, task.cancel(true) (:1018) interrupts the other queues' in-flight flushes, and close() does the same through shutdownNow(). Those workers now hit the InterruptedException branch, lose the complete(error) race, and log "Timer flush failed for table ... after the round ended" with a stack trace, even though the interrupt is the intended stop signal and not a table failure.
• Risk: no correctness impact, but on every failed round or close-during-flush operators see one extra WARN per healthy queue that looks like a second table failure, which partly works against the goal of Issue 2 (making the real secondary failures easy to find).
• Best improvement: skip the WARN (or log at DEBUG without the stack) when error instanceof InterruptedException and the round is already done; keep WARN for genuine late failures.
• Severity: Low
• Raised by another reviewer: No

Status of the points from my previous review

#	Previous issue	Status at d209c351
1	Pre-existing prepareCommit lock-order deadlock under CONTINUE_OTHER_TABLES	Still present and still pre-existing (MultiTableSinkWriter.java:760 holds the queue lock while :785 calls handleTableFailure, which takes every queue lock in removeTableWriters, :1309-1314). Out of scope here; I would still suggest a follow-up issue. Non-blocking.
2	Later child flush errors dropped without logging	Fixed. Losing errors are now logged with table id and queue index (:1040-1046, :1089-1095), including the case where isolation is skipped because the round already ended.
3	cancel(true) interrupts healthy in-flight flushes on FAIL_FAST	Unchanged (:1018). Acceptable as a deliberate prompt-stop choice; it now also produces the WARN noise described in Issue A. Non-blocking.
4	Outdated Javadoc on registerAggregatedFlushIfNeeded / close()	Fixed. MultiTableSink.java:259-265 now says the sink task runs the action on a flush signal and queues flush in parallel; the close() Javadoc (MultiTableSinkWriter.java:855-869) describes stopping the round, the isolation barrier, and the actual IOException/RuntimeException contract.
-	Optional cross-queue FAIL_FAST stop test	Fixed. See 2.2.

2 Quality

2.1 Code standards and comments

The new branch structure in flushQueue is compact and the existing comment on why isolation coordinates with close() outside the queue lock is kept. Javadocs now match the behavior.

2.2 Tests

• testAggregatedFlushReportsFailureAndClosesQueuesIndependently (MultiTableSinkWriterTest.java:633) now puts a second child (test.sibling, index 3) on queue 1 next to the slow one, blocks whichever child is visited first, fails queue 0, and asserts slowQueueFlushCount == 1 (:729). Removing the cross-queue completion.isDone() check in flushQueue would make the second child flush and turn this red, so it closes the gap I mentioned. Using "whichever is visited first" also makes it independent of ConcurrentHashMap iteration order. Stability: Stable (latches, non-interruptible join, bounded waits).
• The other new cases (:577, :733) are unchanged and remain Stable.
• There is no assertion on the new WARN logging, which is fine for log-only behavior.
• All four unit-test jobs (JDK 11/17, ubuntu and windows) passed on this head in the fork run, so the API tests including MultiTableSinkWriterTest are green.

2.3 Docs

docs/en/engines/zeta/timer-flush.md and docs/zh/engines/zeta/timer-flush.md are unchanged since my last review and still consistent with each other and with the code.

3 Architecture

Still the right place: the change stays inside MultiTableSinkWriter, reuses the executor and queue locks that prepareCommit already relies on, keeps per-queue ordering, and leaks no connector-specific logic into the API layer.

4 Issue Summary

#	Issue	Location	Severity	Blocking
A	Expected interrupts logged at WARN with stack as "timer flush failed"	MultiTableSinkWriter.java:1074-1075,1089-1095	Low	No
1	Pre-existing prepareCommit lock-order deadlock (continue-other-tables)	MultiTableSinkWriter.java:760,785,1309	Medium	No (follow-up)
3	cancel(true) interrupts healthy flushes on FAIL_FAST	MultiTableSinkWriter.java:1018	Low	No

5 Merge Recommendation

Ready to merge after fixes

• Non-CI blockers: none. Issue A, 1 and 3 are non-blocking suggestions.
• Remaining blocker is CI only. The fork Build run for this head (https://github.com/nzw921rx/seatunnel/actions/runs/42742314342) has 77 jobs passed, but connector-file-sftp-it failed on JDK 11 and 17 because the atmoz/sftp:alpine-3.7 image no longer exists on Docker Hub (404 when pulling), and a few jobs are still running. That failure is unrelated to this PR and is already fixed on dev by #13203; the branch is 2 commits behind dev, which includes that fix. Could you please merge the latest dev into this branch so CI reruns with the fixed SFTP image? There are no conflicts. I will be glad to approve once CI is green.

Thanks again for turning the feedback around so quickly and for keeping the shutdown and isolation edge cases so tight.