nielifeng opened a new issue, #11471: URL: https://github.com/apache/seatunnel/issues/11471
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue) and found no similar issues. ### What happened When a Kafka topic is written by SeaTunnel Kafka sink with `semantics = EXACTLY_ONCE`, a subsequent Zeta batch job using Kafka source can read all visible user records and write them to the sink, but the job does not finish. The same Kafka source batch job finishes normally when reading: - a topic populated by `kafka-console-producer` - a topic populated by SeaTunnel Kafka sink with `semantics = NON` The reproducible difference is that the transactional topic contains Kafka transaction control record(s). In my validation with 100 user records: - EXACTLY_ONCE topic offsets: `partition 2 end offset = 101` - Console consumer visible records: `100` - Kafka source -> JDBC sink metrics after 90 seconds: - `jobStatus = RUNNING` - `SourceReceivedCount = 100` - `SinkWriteCount = 100` - `SinkCommittedCount = 100` - `IntermediateQueueSize = 0` - target table count = `100` For the control case with `semantics = NON`: - topic end offset = `100` - Console consumer visible records = `100` - Kafka source -> JDBC sink job status = `FINISHED` - target table count = `100` It looks like the batch Kafka source may use the raw end offset including transaction control records as the stopping offset, while the Kafka consumer only emits the 100 user records. As a result, the source has consumed all visible records but never satisfies the batch completion condition. ### SeaTunnel Version `3.0.0-SNAPSHOT`, tested on Zeta engine build commit `46d2f39`. ### SeaTunnel Config Step 1: write a transactional Kafka topic with SeaTunnel Kafka sink: ```hocon env { job.name = "kafka-exact-write" job.mode = "BATCH" parallelism = 1 checkpoint.interval = 10000 checkpoint.timeout = 60000 } source { FakeSource { plugin_output = "fake" auto.increment.enabled = true auto.increment.start = 1 row.num = 100 schema { fields { id = "int" name = "string" age = "int" } primaryKey { name = "pk" columnNames = [id] } } } } sink { kafka { plugin_input = "fake" topic = "zeta_tx_recheck_exact" bootstrap.servers = "kafka:9092" format = json semantics = EXACTLY_ONCE transaction_prefix = "zeta2-tx-recheck-exact" kafka.config = { acks = "all" request.timeout.ms = 60000 buffer.memory = 33554432 } } } ``` Step 2: read that topic with Kafka source in batch mode: ```hocon env { job.name = "kafka-exact-read" job.mode = "BATCH" parallelism = 1 checkpoint.interval = 10000 checkpoint.timeout = 60000 } source { Kafka { plugin_output = "kafka_rows" schema { fields { id = "int" name = "string" age = "int" } primaryKey { name = "pk" columnNames = [id] } } format = json topic = "zeta_tx_recheck_exact" bootstrap.servers = "kafka:9092" start_mode = "earliest" consumer.group = "kafka-exact-read" kafka.config = { client.id = "kafka-exact-read" max.poll.records = 50 auto.offset.reset = "earliest" enable.auto.commit = "false" } } } sink { jdbc { plugin_input = "kafka_rows" url = "jdbc:mysql://mysql:3306/zeta_test" driver = "com.mysql.cj.jdbc.Driver" user = "zeta" password = "******" query = "insert into zeta_tx_recheck_exact(id,name,age) values(?,?,?)" } } ``` Control case: change the Kafka sink in step 1 to `semantics = NON`. The read job finishes normally. ### Running Command Submitted through Zeta REST API: ```shell curl -H 'Content-Type: text/plain' \ --data-binary @job.conf \ 'http://<zeta-master>:8080/submit-job?format=hocon' ``` ### Error Exception No exception is thrown. The job remains `RUNNING` even though all user records have been read and committed to the JDBC sink. Observed job info for the hanging read job: ```json { "jobStatus": "RUNNING", "errorMsg": null, "metrics": { "SourceReceivedCount": "100", "SinkWriteCount": "100", "SinkCommittedCount": "100", "IntermediateQueueSize": "0" } } ``` ### Zeta or Flink or Spark Version Zeta engine. ### Java or Scala Version Java 11. ### Screenshots N/A ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### 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]
