zephyr45 commented on PR #13060: URL: https://github.com/apache/seatunnel/pull/13060#issuecomment-6580809968
@SEZ9 Thanks for the concrete final-pass questions. I reproduced the additional Flink counterexample before changing production code, and pushed the coherent fix in `97516b977`. Here is the requested ledger. 1. **F1 — ownership rule, supported engines, and repeated rescaling.** The connector still has no modulo/index ownership rule; it consumes exactly the states assigned by the engine. Existing regressions cover Zeta-style modulo and Flink-style balanced contiguous assignments. The new `3 -> 4 -> 3 -> 4 -> 3` regression models the common Flink adapter's writer-local counter restart. On the previous logic it failed at the final 4-to-3 restore with duplicate namespace `shared-prefix-7-gtestw3r2c1`. The fix replaces checkpoint-qualified relocation with a checkpoint-independent recovery revision: `<base>_r<revision>w<current-index>`. Same-recovery retries derive the same namespace; a later recovery sees and advances the checkpointed revision. `FlinkSinkWriterTest` also records that fresh/restored writers legitimately have divergent counters. 2. **F2 — mixed generated prefixes.** Every `KafkaSinkState` is parsed relative to its own saved `transactionIdPrefix`; no common-prefix assumption was reintroduced. `shouldRestoreMixedGeneratedPrefixesAfterScaleDown` remains green, along with legacy/current mixed-path coverage. Malformed or overflowing namespace revisions fail before producer construction. 3. **F3 — contiguous sequences and stateless retry cleanup.** Checkpoint IDs are no longer used for namespace identity or transaction sequence. Each namespace uses contiguous transaction numbers. Assigned old namespaces are probed from saved sequence `n + 1`, preserving pending checkpoint transaction `n`; the deterministic output namespace is probed from 1, so a stateless/recovery retry fences every transaction abandoned after the retained checkpoint. Coverage includes checkpoint jumps, omitted configured prefix, multiple abandoned IDs, and pending-transaction preservation. 4. **F4 — Spark retry identity and evidence boundary.** Spark 2.4/3.3 use stable `partitionId`, not attempt-specific `taskId`, for the shared `SinkWriter.Context` index; the generation remains serialized in `KafkaSink`. The shared change affects all Spark sinks and is called out in the PR body. `SparkSinkTest` verifies the factory path; broker-backed parallel E2E passed on Spark 2.4.6 and 3.3.0. I am **not** describing that batch E2E as a broker-backed Spark crash/retry test. 5. **F5/F6 — concurrent legacy cleanup (one concern).** This is not marked fixed. Multiple restoring tasks may probe the same released legacy prefix, increasing first-upgrade broker round-trips roughly quadratically while preserving correctness. Local map deduplication cannot coordinate tasks, and subtask 0 would miss historical generated prefixes assigned elsewhere. A safe optimization needs per-prefix cross-task ownership plus owner failover. Please explicitly confirm whether this low-severity startup optimization can remain a documented follow-up; this patch does not weaken cleanup to hide it. 6. **F7 — documentation and upgrade boundary.** EN/ZH option rows retain the public opaque shape `<prefix>-<writer>-<sequence>`. Semantics document Zeta/Flink recovery scope, checkpoint-independent revisions, legacy fencing, and `spark.speculation=false`. Both incompatible-change pages require stopping the old attempt, rebuilding the execution graph from upgraded job configuration, restoring retained checkpoint/savepoint state, updating ACLs, and retaining legacy-ID access during the first restore. `KafkaSinkState`, configuration, public API/SPI, and dependencies remain unchanged; old serialized `KafkaSink` execution graphs are explicitly not supported. 7. **Additional Flink finding and cleanup hardening.** The handoff counterexample was valid as a source-level/adaptor model, not a prior full Flink rescale deployment reproduction. The new recovery-revision invariant fixes it without extending engine APIs. Cleanup now also closes temporary producers when `initTransactions()` or `beginTransaction()` fails and preserves the original exception (close failures are suppressed), with two regressions. Verification for `97516b977f7d65ef2096ac53376e1460ee8f61d4` content: - Kafka focused: 27/27; full connector: 152/152. - Flink common: 15/15; focused adapter regression: 4/4. - Spark 3.3 unit: 2/2; Spark 2.4 package: pass (no tests). - Kafka 7.0.9 broker E2E: Zeta 2/2 (parallel + 3-to-2 restore), Flink 4/4 (1.13.6/1.15.3/1.18.0/1.20.1 parallel), Spark 2/2 (2.4.6/3.3.0 parallel). - Docs: 7 executed, 0 failures/errors, 3 conditional skips; `git diff --check` and scoped Spotless pass. - Repository-wide Java 11 `verify -DskipTests`: pass. Exact-head GitHub CI is now running; I will post the final result and investigate/rerun only genuine failed jobs. The PR description has been rewritten around the final invariant, migration procedure, shared Spark impact, evidence boundaries, and the explicit legacy-cleanup follow-up. @SEZ9, please re-review this head and clear the outstanding change request if the answers and implementation are satisfactory. Your collaborator review is also the maintainer-side action still required; I am not requesting or performing a merge. -- 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]
