nzw921rx commented on PR #10306: URL: https://github.com/apache/seatunnel/pull/10306#issuecomment-5113753690
# Deep Review — Error Data Bypass / DLQ Thanks for continuing to iterate on this feature. I reviewed the current head `2a56e4fdbdb617e4a66d3ef305a20b6334646e67` and rechecked the existing 71 discussion entries and 10 review threads to avoid repeating previously reported issues. The findings below are introduced by the current PR changes and are not duplicates of the existing feedback. ## Review conclusion I do not think the current revision is ready to merge. I found two merge-blocking silent data-loss/silent-success regressions, two MultiTable correctness gaps, and one checkpoint-metrics attribution issue. ## 1. Blocker: successful JDBC batches can be rolled back and silently lost with `auto_commit=false` **Location** * [`[JdbcSinkWriter.java:209-223](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java#L209-L223)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java#L209-L223) * [`[JdbcSinkWriter.java:364-423](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java#L364-L423)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java#L364-L423) `clearPendingRowsIfAutoFlushed()` now clears `pendingRows` after every successful `executeBatch`. However, with `auto_commit=false`, an auto-flushed batch is still part of the current uncommitted transaction. A deterministic failure sequence is: 1. Configure `batch_size=2` and `auto_commit=false`. 2. Valid rows A1/A2 trigger a successful auto-flush. 3. A1/A2 are removed from `pendingRows`, although the transaction has not been committed. 4. Rows B1/B2 later trigger a SQLState 22/23 batch failure. 5. `rollbackIfNeeded()` rolls back the entire transaction, including A1/A2. 6. Only B1/B2 are passed to the error collector because A1/A2 have already been forgotten. 7. The engine-side deferred outcome queue can subsequently report A1/A2 as `WRITTEN`. The result is silent data loss: A1/A2 exist in neither the main sink nor ErrorData, while metrics and stain tracing can still report them as successful. This regression was introduced when the condition changed from `autoCommit && autoFlushed` to `autoFlushed`. Please retain transaction-scoped rows until commit, or otherwise ensure that every row affected by a rollback is reported consistently. A regression test should cover a successful auto-flushed batch followed by a row-level failure in the same non-auto-commit transaction. ## 2. Blocker: SQL query-level failures are classified as row errors and can be silently swallowed **Location** * [`[SQLTransform.java:122-133](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java#L122-L133)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java#L122-L133) * [`[ZetaSQLEngine.java:292-300](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java#L292-L300)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java#L292-L300) * [`[ZetaSQLFilter.java:110-125](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java#L110-L125)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFilter.java#L110-L125) The classifier treats every outer `WHERE_STATEMENT_ERROR` as a row-level error. However, `ZetaSQLEngine` wraps every exception thrown by `executeFilter()` with that same error code, including unsupported query expressions and other query-level failures. For example: ```sql SELECT * FROM t WHERE id BETWEEN 1 AND 5 ``` `BETWEEN` reaches the existing `UNSUPPORTED_OPERATION` branch in `ZetaSQLFilter`, but that exception is wrapped as `WHERE_STATEMENT_ERROR`. With `transform_error_handler=LOG` or `ROUTE`, the new classifier marks it as `ROW_ERROR`, and the wrapper drops or routes every input row. Because the default error thresholds are disabled, the job can finish successfully with zero business output. The pre-existing lack of `BETWEEN` support is not the issue being raised here. The PR-introduced regression is that a query/system failure that previously failed the task is now silently converted into per-row error handling. Please inspect the nested cause or classify exceptions closer to their origin so that unsupported SQL, configuration errors, UDF failures, and system failures remain fail-fast. The tests should execute a real unsupported expression through the full wrapper path rather than only constructing a synthetic outer `WHERE_STATEMENT_ERROR`. ## 3. High: MultiTable terminal-outcome handling has a lost-update race **Location** * [`[EngineRowErrorCollector.java:53-62](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/EngineRowErrorCollector.java#L53-L62)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/EngineRowErrorCollector.java#L53-L62) * [`[MultiTableWriterRunnable.java:129-132](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableWriterRunnable.java#L129-L132)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableWriterRunnable.java#L129-L132) * [`[SinkFlowLifeCycle.java:608-670](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java#L608-L670)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java#L608-L670) When JDBC handles a batch row error internally, the following ordering occurs: 1. `EngineRowErrorCollector.collect()` publishes `ROUTED_TO_ERROR_SINK` or `DROPPED`. 2. `JdbcSinkWriter.write()` returns normally. 3. `MultiTableWriterRunnable` unconditionally calls `writeSuccessHandler.accept(row)`. The task thread drains collector outcomes independently from the MultiTable worker. If it drains after step 1 but before step 3, `removePendingTerminalWriteRow()` cannot find the triggering row because its success callback has not run yet. The worker then adds that row to `pendingTerminalWriteRows` as `WRITTEN`, but the corresponding error outcome has already been consumed. The next timer flush or checkpoint therefore records the same row as both: * `ROUTED_TO_ERROR_SINK`/`DROPPED`; and * `WRITTEN`/`SINK_WRITE_DONE`. The collector list and pending-success deque use separate locks, so there is no atomic ordering between these operations. Please publish success/error terminal states through one synchronization domain, or make the connector return an explicit terminal outcome instead of treating every normal return as success. A deterministic test can use latches to pause the worker between collector publication and `writeSuccessHandler.accept()`. ## 4. High: `CONTINUE_OTHER_TABLES` swallows ErrorHandler’s global fail-job policies **Location** * [`[ErrorHandler.java:154-217](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/ErrorHandler.java#L154-L217)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/error/ErrorHandler.java#L154-L217) * [`[MultiTableWriterRunnable.java:209-229](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableWriterRunnable.java#L209-L229)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableWriterRunnable.java#L209-L229) * [`[MultiTableSinkWriter.java:822-844](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java#L822-L844)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java#L822-L844) `ErrorHandler` deliberately throws when: * `max_error_records` is exceeded; * `max_error_ratio` is exceeded; * the ErrorData sink fails; or * `queue_overflow_policy=FAIL` is triggered. When such an exception originates inside `JdbcSinkWriter`, it is propagated as an `IOException`. Under `CONTINUE_OTHER_TABLES`, `MultiTableWriterRunnable` treats it as an ordinary table-scoped write failure and quarantines only the current table. This means a streaming job can continue running even after: * the stage-wide error budget has been exceeded; * the shared ErrorData destination has failed; or * the configured queue policy explicitly requested job failure. These are action/stage-level failures, not table-local failures. In particular, continuing after the DLQ itself has failed can silently discard subsequent error records. Please preserve a distinguishable fatal error type and bypass table isolation for ErrorHandler threshold, queue, and ErrorData-sink failures. The same distinction is needed in runtime-write, checkpoint, and timer-flush paths. ## 5. Medium: deferred success metrics are assigned to the next checkpoint **Location** * [`[SinkFlowLifeCycle.java:723-736](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java#L723-L736)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java#L723-L736) * [`[ConnectorMetricsCalcContext.java:228-253](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/metrics/ConnectorMetricsCalcContext.java#L228-L253)`](https://github.com/apache/seatunnel/blob/2a56e4fdbdb617e4a66d3ef305a20b6334646e67/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/metrics/ConnectorMetricsCalcContext.java#L228-L253) Checkpoint N calls `sealCheckpointMetrics()` before `prepareCommit()`. The new deferred `WRITTEN` outcomes are only drained and recorded after `prepareCommit()` returns. `recordTerminalWriteOutcome()` therefore adds these rows to the new `currentPendingMetrics` bucket after checkpoint N has already been sealed. They can only be sealed by checkpoint N+1. Consequences: * data committed by checkpoint N is reported as committed by checkpoint N+1; * if N is the final checkpoint, the committed metrics are never published; * if N+1 aborts, successfully committed data from N can remain absent from `SINK_COMMITTED_COUNT` and `SINK_COMMITTED_BYTES`. Please resolve deferred terminal outcomes before sealing checkpoint metrics, or explicitly associate the resolved outcomes with checkpoint N. ## Suggested regression coverage Before merging, I recommend adding focused tests for: 1. `auto_commit=false`: successful auto-flush followed by a row-level batch failure and rollback; 2. SQL transform: an unsupported query expression must remain a system-level failure; 3. MultiTable: collector publication racing with the success callback; 4. `CONTINUE_OTHER_TABLES`: threshold and ErrorData-sink failures must still fail the job; 5. checkpoint N: deferred successes must be included in checkpoint N’s committed metrics. -- 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]
