nzw921rx opened a new issue, #13044: URL: https://github.com/apache/seatunnel/issues/13044
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues) and found no similar issues. The closest existing reports are closed and cover different failure modes: #4247 (opening transaction), #4949 (invalid transactional state), #5294 (producer close), #10061 / #11534 (checkpoint/restore). None reports the `parallelism > 1` transactionalId collision below. ### What happened A Kafka sink with `semantics = "EXACTLY_ONCE"` and `parallelism = 2` fails with `ProducerFencedException`: ```text org.apache.seatunnel.connectors.seatunnel.kafka.exception.KafkaConnectorException: Kafka transaction [st310-eo-tx-1] failed to send one or more of its 50 record(s) asynchronously. Caused by: org.apache.kafka.common.errors.ProducerFencedException: Producer with transactionalId 'st310-eo-tx-1' and ProducerIdAndEpoch(producerId=4, epoch=17) has been fenced by another producer with the same transactionalId ``` The job FAILED. The test topic ended with 100 unique data records plus 2 transaction control markers; no duplicate data was observed in this run, but the job cannot complete. The same job with `parallelism = 1` succeeds: 100 rows written, topic end offset 101 (one control marker), 100 unique records. This was reproduced on SeaTunnel 3.1.0-SNAPSHOT (latest dev, runtime commit `c8e7896`) on a Zeta 2-master/3-worker cluster with Kafka 3.9.2 (single KRaft broker). The same `prefix + "-" + checkpointId` helper exists in 2.3.13, so this may predate 3.0; it was not bisected. ### Expected behavior Kafka sink `EXACTLY_ONCE` works with `parallelism > 1`. Each sink writer/subtask must use a distinct transactionalId; no writer may fence another writer of the same job. The job should finish and the topic should contain exactly the 100 source records. ### Steps to reproduce 1. Start a Kafka broker with transaction support (single-node KRaft is enough). 2. Run a job with a Kafka sink configured as: ```hocon env { job.mode = "BATCH" parallelism = 2 } source { FakeSource { plugin_output = "fake" row.num = 100 schema { fields { id = bigint, name = string } } } } sink { Kafka { plugin_input = "fake" topic = "st310-eo" bootstrap.servers = "kafka:9092" format = "json" semantics = "EXACTLY_ONCE" transaction_prefix = "st310-eo-tx" } } ``` 3. Submit the job. With `parallelism = 2`, it fails with `ProducerFencedException`; with `parallelism = 1`, it succeeds. ### Root cause (code pointers) `connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java`: - lines 88 and 107 start a transaction with `generateTransactionId(this.transactionPrefix, this.lastCheckpointId + 1)`; - lines 164-165 implement: ```java protected static String generateTransactionId(String transactionPrefix, long checkpointId) { return transactionPrefix + "-" + checkpointId; } ``` The subtask index is not part of the id. All parallel writers therefore use the same transactionalId (e.g. `st310-eo-tx-1`), and Kafka fences all but one. `SinkWriter.Context` already exposes `getIndexOfSubtask()` (`seatunnel-api/.../sink/SinkWriter.java:106`), so the writer index is available. ### Suggested fix Include the writer/subtask index in the transactionalId, for example: ```java transactionPrefix + "-" + context.getIndexOfSubtask() + "-" + checkpointId ``` Ensure the same formula is used for restore/abort paths (`snapshotState`, `abortTransaction`, state restore), and add: - a unit test with two writers sharing a prefix, asserting the generated transaction ids differ; - an E2E Kafka sink test with `semantics = EXACTLY_ONCE`, `parallelism = 2`, and 100 records, asserting the job succeeds without `ProducerFencedException`. ### Workaround Use `parallelism = 1` for the Kafka sink under `EXACTLY_ONCE`, or use `AT_LEAST_ONCE` when parallel writes are required. ### Environment - SeaTunnel: 3.1.0-SNAPSHOT (latest dev, runtime commit `c8e7896`), Zeta 2 masters / 3 workers - Kafka: 3.9.2, single KRaft broker - Source: MySQL 8.0.46 / 100 rows for the E2E run (the minimal FakeSource config above expresses the same sink setup) - Sink format: JSON ### Are you willing to submit a PR? Yes — I can help test and review a fix; an implementation PR from the community is welcome (hence `help wanted`). ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
