DanielLeens commented on PR #10306: URL: https://github.com/apache/seatunnel/pull/10306#issuecomment-5181738334
## Review: Error Data Bypass Capability for Transform & Sink (Dead Letter Queue) Source-level review at head `3da1fc962d`. This is a large, well-tested feature (124 files, +14k lines): env-configured row-level error handling (`DISABLE`/`LOG`/`ROUTE`) wired into Zeta's `TransformFlowLifeCycle`/`SinkFlowLifeCycle`, an async `DefaultErrorSinkWriter` with its own plugin classloader, Hazelcast-backed stage-wide thresholds, JDBC SQLState-based classification (`22xxx`/`23xxx`), MultiTableSink integration with deferred terminal outcomes, and bilingual docs. Test breadth is genuinely impressive — ~40 unit tests plus two ITs covering all modes, overflow policies, thresholds, and failure paths, and the one touched legacy assertion (`MultiTableSinkWriterTest` lock probe) was made awaitility-based without losing strictness. That said, I found several issues that I believe should be addressed before merge. Ordered by severity: ### Blocking / high priority **1. The default (unconfigured) JDBC path is NOT unchanged — and one combination silently loses data.** The PR description says "If it is not configured, the execution logic will be the same as before", but two changes apply regardless of any `error_handler` config: - `JdbcOutputFormat.flush()` now throws immediately on SQLState `22xxx`/`23xxx` instead of retrying up to `max_retries`. Defensible (these are deterministic), but it is a default-behavior change for every existing JDBC job and belongs in the docs / `incompatible-changes.md`. - `JdbcSinkWriter.write()` (non-collector branch) now calls `outputFormat.clearBatchSilently()` on a row-level error before rethrowing. Under the pre-existing MultiTableSink continue-on-failure policy, `MultiTableWriterRunnable.writeWithRetry()` then re-invokes `write(row)` for the current row only — every previously buffered row of the discarded batch is silently lost, the table is never quarantined, and the job completes "successfully". Before this PR the same scenario ended in a visible table failure. This converts a visible failure into silent partial data loss on a path where the user has not opted into anything. Suggestion: only clear the batch when a `RowErrorCollector` is present (i.e., the user opted in), or quarantine the table after clearing so the failure stays visible. **2. `MultiTableState` serialization compatibility must be proven.** The class previously had no explicit `serialVersionUID`; the PR adds `serialVersionUID = 5992121739651030596L` together with the new `failedTables` field, and this state is persisted via `DefaultSerializer` (Java serialization) into checkpoints/savepoints. If that constant is not exactly the `serialver` of the pre-PR class, every restore from a checkpoint taken on an older version fails with `InvalidClassException`. The null-tolerant `getFailedTables()` suggests this was considered — please confirm in the PR description how the value was generated, and ideally add a test that deserializes a state blob captured from the old class shape. **3. Unbounded memory growth in `EngineRowErrorCollector` on the MultiTableSink + deferred-outcome path.** In the common flow, the worker thread calls `consumeCollectedRowErrorOutcome(row)` immediately after `write(row)` returns (row merely buffered — no outcome yet), and the batch's `collectWriteSuccess` outcomes are drained later by the task thread via `drainTerminalOutcomes(rememberRecordedRows=true)`, which puts every such row into `recordedTerminalRows`. Nothing ever consumes those entries (the worker's one-shot consume already happened), and there is no periodic cleanup — the `IdentityHashMap` retains every successfully flushed `SeaTunnelRow` for the life of the task. On a streaming job this is an OOM. The `terminalOutcomes` list / `pendingTerminalWriteRows` deque are also scanned linearly per outcome, which is O(n²) as batches grow. **4. Documentation contradicts the implementation on threshold counters — and the implemented behavior is a per-row distributed hot-path cost.** `error-handling.md` states: "Row processing updates local task counters first and publishes their deltas to shared engine state only after the corresponding checkpoint completes, so the hot path does not perform a shared-state write per row and aborted checkpoints do not advance committed thresholds." The code does the opposite: `StateStoreErrorHandlerCounter.incrementTotalRecords()` executes a Hazelcast `IMap.compute` per row, plus an `IMap.get` for the ratio check argument — i.e., two distributed operations for every record flowing through an error-enabled stage — and its `snapshotState`/`notifyCheckpointComplete`/`notifyCheckpointAborted` are no-ops (the `ErrorHandler` javadoc also describes the checkpoint-delta design that isn't implemented). This will cap throughput severely once `mode != DISABLE`. Either implement the documented local-counter + checkpoint-delta design (the lifecycle hooks are already plumbed through both flow lifecycles), or fix the docs and clearly warn about the per-row distributed cost. **5. `MultiTableSinkWriter.removeTableWriters()` regression risk for the existing continue-on-failure feature.** Previously a quarantined writer whose `close()` failed was logged (`log.warn`) and removed; now the close error is rethrown, which propagates through `registerFailedTable`/`handleTableFailure` (or poisons the worker via `notifyTableFailure`) and fails the job. A writer typically gets quarantined precisely because its connection is broken, and closing a broken JDBC writer frequently throws — so the "continue other tables" policy now fails the whole job in one of its most common triggering scenarios. If keeping half-closed writers around is the concern, consider logging + best-effort removal instead of failing the job. ### Medium - **Counter state is never cleaned up.** Keys in `engine_error-handler-counter-map` (`v1:jobId:pipelineId:actionId:stage:*`) are created via `initializeIfAbsent` but never removed on job completion/cancellation — unbounded IMap growth on long-lived clusters. Also note replayed rows after recovery increment `total` again, so ratio denominators drift across restarts (counter reuse is documented, the replay skew is not). - **Env options are parsed by hand instead of the `Option` framework.** `ErrorHandlerConfigUtil` reads raw nested maps (`error_handler`, `sink_error_handler`, ...). Project convention is that user-facing configs are `Option`-defined (stable contract, typed validation, doc generation). At minimum the keys should be defined as constants; the unused `jobId` parameter in `buildStageConfig` can go. - **Batch-granularity DLQ semantics with autocommit.** With `autoCommit=true` (the default), a partially applied batch is not rolled back, yet all rows of the batch are routed to the error table — rows can exist in both the target and the error table, so replaying the error table duplicates data. The batch-coarseness itself is documented well; this duplicate-on-backfill hazard deserves an explicit doc warning. The `TODO achieves precise error data filtering` (per-row fallback execution, as Flink's JDBC sink does) would resolve both. - **Hardcoded 60s stalls.** Checkpoint barriers now block on `waitForPendingRows` (error queue drain) with a hardcoded 60s limit; `MultiTableSinkWriter.close()` gained a 60s `awaitTermination` that converts slow shutdown into job failure; `DefaultErrorSinkWriter.close()` interrupts the worker before draining, so error rows enqueued during close can be dropped or produce spurious close failures if the interrupt lands mid-JDBC-write. Consider making the timeouts configurable and draining before interrupting. - **Connector-specific code in the engine.** `DefaultErrorSinkWriter.shutdownMySqlAbandonedConnectionCleanupThreadIfPresent()` reflects on `com.mysql.cj.jdbc.AbandonedConnectionCleanupThread` inside `seatunnel-engine-server`. Understandable as a classloader-leak fix, but it's exactly the connector-logic-in-engine leakage the architecture guidelines warn about. Relatedly, whitelisting `seatunnel-error-sink-*` in the e2e thread-leak check masks future leaks of these very threads. - **Transform ratio semantics.** Each operator in a transform chain increments the shared total per row (documented), so `max_error_ratio` is diluted by chain length — a 3-transform chain needs 3x the error rate to trip the same ratio. Worth reconsidering counting once per stage input row. ### Minor / nits - `seatunnel-e2e/.../seatunnel-transforms-v2-e2e-part-2/src/test/resources/multi_table_merge.conf` is not referenced by any test (`TestTableMergeIT` runs `table_merge_multi_table.conf`) — orphaned file, please remove or wire it up. - Unrelated formatting-only edits should be dropped per the one-PR-one-problem rule: the three `seatunnel-engine-ui` files (trailing newline only), `docs/zh/engines/zeta/realtime-observability.md` and `rest-api-job-lifecycle.md` (trailing whitespace only), and `batch_fake_to_console_with_transform_name.conf` (trailing newline). - Dead code: `SinkFlowLifeCycle#getCollectedErrorCount/getCollectedRoutedCount/getCollectedDroppedCount` are unused; `rollbackIfNeeded()` has a pointless catch-and-rethrow; `isRowLevelSqlState` is duplicated between `JdbcOutputFormat` and `JdbcSinkWriter` (extract a shared helper). - `DefaultErrorSinkWriter.ensureInitialized()` is a synchronized method invoked on every error write — a volatile fast-path check would avoid the monitor on the steady state. - `ErrorHandler` logs a WARN (optionally with stacktrace) per error row in LOG/ROUTE mode — at high error rates this floods logs; consider sampling/rate-limiting. - Wrapping transforms in `ErrorHandling(Flat)MapTransform` hides any other capability interfaces the delegate implements from downstream `instanceof` checks — currently benign, but worth a defensive comment. ### Test coverage gaps - No restore/recovery E2E with error handling enabled (the `MultiTableState.failedTables` restore path and counter reuse after recovery are unit-tested only). - No test covers the two default-path behavior changes in issue 1 (retry skip, batch clear without collector under table-retry). - Checklist: `incompatible-changes.md` was not updated despite the default-path changes above. ### Verdict The architecture is sound (opt-in wrapping, classifier interfaces, isolated error-sink lifecycle validation, honest experimental framing in the docs), and the test investment is far above average. But issues 1–5 — silent data loss on an unconfigured path, checkpoint-restore compatibility proof, a streaming memory leak, a doc-vs-code contradiction on a hot-path distributed write, and a continue-on-failure regression — should be resolved or explicitly justified before this merges. Happy to re-review after updates. *Note: this review is based on reading the source at the current head; I did not run builds or the E2E suite locally — CI results remain authoritative for those.* -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
