nielifeng opened a new issue, #11472: URL: https://github.com/apache/seatunnel/issues/11472
### 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 After a Zeta worker failover, a batch job can finish successfully and the target data can be correct, but the final job metrics returned by REST `/job-info/{jobId}` are inconsistent with the actual bounded input and target table. I reproduced this with a deterministic Kafka topic containing exactly `200000` JSON records with `id = 1..200000`, then ran a Zeta batch job: ```text Kafka source -> JDBC MySQL sink ``` The sink table has `id` as primary key and uses `insert ignore` so retries are idempotent. This lets us validate data loss by checking count/distinct/min/max/gaps. No-failover baseline: ```text jobStatus = FINISHED SourceReceivedCount = 200000 SinkWriteCount = 200000 SinkCommittedCount = 200000 IntermediateQueueSize = 0 MySQL: count = 200000 distinct_ids = 200000 min_id = 1 max_id = 200000 missing_in_range = 0 ``` Worker failover run 1: ```text jobStatus = FINISHED errorMsg = null SourceReceivedCount = 303148 SinkWriteCount = 299048 SinkCommittedCount = 200000 IntermediateQueueSize = 4098 MySQL: count = 200000 distinct_ids = 200000 min_id = 1 max_id = 200000 missing_in_range = 0 ``` Worker failover run 2: ```text jobStatus = FINISHED errorMsg = null SourceReceivedCount = 239640 SinkWriteCount = 238024 SinkCommittedCount = 200000 IntermediateQueueSize = 1616 MySQL: count = 200000 distinct_ids = 200000 min_id = 1 max_id = 200000 missing_in_range = 0 ``` So the data result is correct after failover, but `SourceReceivedCount`, `SinkWriteCount`, and `IntermediateQueueSize` in the finished job metrics appear to include records from canceled/retried task attempts, or are not reset/reconciled when the replacement attempt finishes. This makes finished job metrics misleading for reliability validation and monitoring. In particular, `IntermediateQueueSize` should not remain non-zero after a batch job is `FINISHED`. ### SeaTunnel Version `3.0.0-SNAPSHOT`, tested on Zeta engine build commit `46d2f39`. ### SeaTunnel Config The source Kafka topic was pre-populated with exactly `200000` JSON records: ```json {"id":1,"name":"n1","age":21} ... {"id":200000,"name":"n200000","age":20} ``` Kafka offsets and visible records were both verified as `200000`. Job config: ```hocon env { job.name = "zeta-kafka-deterministic-workerfail-200k" job.mode = "BATCH" parallelism = 2 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_kafka_deterministic_200k" bootstrap.servers = "kafka:9092" start_mode = "earliest" consumer.group = "zeta-kafka-deterministic-workerfail-200k" kafka.config = { client.id = "zeta-kafka-deterministic-workerfail-200k" max.poll.records = 500 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 ignore into zeta_kafka_det_workerfail_200k(id,name,age) values(?,?,?)" } } ``` Failover action: ```shell systemctl stop seatunnel-worker # wait until the batch job finishes systemctl start seatunnel-worker ``` ### 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' ``` Metrics checked through: ```shell curl 'http://<zeta-master>:8080/job-info/<jobId>' ``` ### Error Exception No exception is thrown. The job status is `FINISHED` and target data is correct. The problem is inaccurate finished job metrics after worker failover: ```json { "jobStatus": "FINISHED", "errorMsg": null, "metrics": { "SourceReceivedCount": "303148", "SinkWriteCount": "299048", "SinkCommittedCount": "200000", "IntermediateQueueSize": "4098" } } ``` ### Zeta or Flink or Spark Version Zeta engine, 3 masters and 2 workers. Each worker has 8 fixed slots. ### 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]
