zephyr45 commented on PR #13060: URL: https://github.com/apache/seatunnel/pull/13060#issuecomment-6594515644
@SEZ9 Thanks - the earlier notification evidently truncated the ledger, so here is a self-contained answer against the current head `b9ed7c1297d9` (which contains the recovery-revision commit `97516b977` plus the bounded fresh-writer follow-up). 1. **F1 - engine-neutral ownership and a true red/green.** The connector consumes the states assigned by the engine; there is no modulo/index ownership validation. The regression is [`shouldKeepNamespacesDistinctWhenFlinkLocalCheckpointCountersRestart`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriterTest.java#L336-L382), which models `3 -> 4 -> 3 -> 4 -> 3` with empty-state Flink writers restarting their local checkpoint counter at 1. I re-ran the same test method in isolated detached worktrees on Java 11: - Pre-fix production logic at `351106062`, with only that exact regression method copied in: **1 test, 1 failure** at the final 4-to-3 restore: `Duplicate namespace at parallelism 3: shared-prefix-7-gtestw3r2c1`. - Exact fix `97516b977`: **1 test, 0 failures**. - Current head `b9ed7c129`: **1 test, 0 failures**. The adapter premise is separately recorded by [`FlinkSinkWriterTest.testFreshWriterCounterRestartsIndependentlyAfterScaleUp`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/sink/FlinkSinkWriterTest.java#L92-L111). **Revision lifecycle:** [`restoreState`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L243-L285) parses every assigned `KafkaSinkState`, selects one writer-specific lineage, and calls `advanceWriterNamespace`. [The parser](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L362-L427) reads the base and optional revision from the checkpointed transaction ID. [`advanceWriterNamespace`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L299-L317) uses `Math.addExact(oldRevision, 1)` and encodes `<bas e>_r<nextRevision>w<currentIndex>`. The writer opens a transaction under that prefix, and [`KafkaTransactionSender.snapshotState`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaTransactionSender.java#L220-L229) writes the complete transaction ID (therefore the revision) back into `KafkaSinkState`. The next recovery reads that same ID through the parser and advances again. `KafkaSinkState` fields/serialization are unchanged. 2. **F2 - mixed generated prefixes.** Each state is parsed relative to its own saved `transactionIdPrefix`; no common-prefix assumption is made. The writer then deterministically adopts one non-legacy state actually assigned to it. [`shouldRestoreMixedGeneratedPrefixesAfterScaleDown`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriterTest.java#L226-L252) covers mixed historical per-writer generated prefixes. Malformed, foreign, or overflowing revisions fail before producer creation. 3. **F3 - scale-up writer with no state.** Yes, a fresh writer starts at sequence 1 independently of the engine checkpoint ID ([fresh restore plan](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L243-L255)). Opening `<namespace>-1` immediately fences an earlier holder of ID 1. Current head intentionally does not walk all later historical IDs for a stateless writer ([cleanup boundary](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java#L319-L349)); a later ID abandoned before any state was retained is resolved by Kafka's `transaction.timeout.ms`. [The unit regression](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-co nnectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriterTest.java#L438-L464) proves repeated fresh construction initializes only ID 1 without extending history. 4. **F4 - Spark retry identity.** Spark 2.4 [uses stable `partitionId`](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-2.4/src/main/java/org/apache/seatunnel/translation/spark/sink/writer/SparkDataWriterFactory.java#L65-L70), and Spark 3.3 [does the same](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/sink/write/SeaTunnelSparkDataWriterFactory.java#L67-L71); retry-specific `taskId` is no longer the sink-writer index. The sink-level generation is serialized, so a retry of one partition reuses the same base namespace and ID 1 fences the previous attempt. `SparkSinkTest` verifies `partitionId=2, taskId=99` produces writer index 2. The shared behavior for Kafka, JDBC XA, Doris, SelectDB, MQTT, File, Pai mon, NATS JetStream, and HugeGraph is now [documented as incompatible behavior](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/docs/en/introduction/concepts/incompatible-changes.md#L21-L23). 5. **F5/F6 - legacy shared-prefix fencing.** Repeated cross-writer probing on the first restore from released shared-prefix state remains present. Correctness is preserved: every assigned legacy state is fenced, and cleanup is deduplicated within a writer. Avoiding the roughly quadratic startup round trips safely requires per-prefix cross-task ownership, owner failover, and mixed-prefix handling; choosing subtask 0 would miss prefixes assigned elsewhere. This is the explicit documented optimization follow-up, and DanielLeens also accepted it as non-blocking. 6. **F7 - option row and engine scope.** The EN [`transaction_prefix` option row](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/docs/en/connectors/sink/Kafka.md#L60-L67) states `<prefix>-<writer>-<sequence>`; ZH has the matching row. The [Semantics section](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/docs/en/connectors/sink/Kafka.md#L94-L108) scopes stateful recovery/rescaling to Zeta and Flink, documents revision advancement, legacy fencing, the stateless timeout boundary, and separately requires `spark.speculation=false` for Spark. EN/ZH incompatible-change pages match. 7. **Current-head broker evidence.** `b9ed7c12` adds [bounded fresh-writer initialization](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java#L2350-L2398) and [abandoned-transaction restore fencing](https://github.com/zephyr45/seatunnel/blob/b9ed7c1297d9e1ca517f171c688641363a9ab912/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java#L2400-L2464). Both passed locally against the pinned real Kafka 7.0.9 broker (2/2); full Kafka units passed 152/152. Exact-head [Build 41561178312](https://github.com/zephyr45/seatunnel/actions/runs/41561178312) is still active, so I am not claiming final CI here. I will handle only justified failed-job reruns after it finishes and post the exact final result separately. -- 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]
