nielifeng opened a new issue, #11497: URL: https://github.com/apache/seatunnel/issues/11497
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues) and found no similar issues. ### What happened The Zeta REST job metrics can report a negative `IntermediateQueueSize` while a streaming job is running normally. During a 72-hour fixed-slot streaming soak, `IntermediateQueueSize` was sampled every 15 seconds from the `running-jobs` REST endpoint. The metric reported `-1` or `-2` in 111 samples: - segment 1: 58 negative samples out of 8,369 post-warm-up samples - segment 2: 53 negative samples out of 8,381 post-warm-up samples The negative values were isolated single samples. The job stayed `RUNNING`, source and sink QPS remained aligned, checkpoints continued, and slot allocation did not change. This appears to be a metrics correctness issue rather than an execution failure. ### SeaTunnel Version `3.0.0-SNAPSHOT`, commit `46d2f3909f7e97f8f516c8fba68d12df471d7718`. ### SeaTunnel Config Relevant workload properties: ```hocon env { job.mode = "STREAMING" parallelism = 10 checkpoint.interval = 30000 checkpoint.timeout = 120000 } source { FakeSource { split.read-interval = 1 string.length = 128 } } sink { Console { log.print.data = false } } ``` The cluster used fixed slots with `WAIT` scheduling. The job occupied 11 slots. ### Running Command Submit the streaming job and repeatedly sample: ```bash curl http://<active-master>:8080/running-jobs ``` At sufficiently high throughput, inspect `metrics.IntermediateQueueSize`. The issue is intermittent, so a longer or higher-frequency sample is more likely to reproduce it. ### Error Exception Example metric values returned by REST: ```text "IntermediateQueueSize":"-1" "IntermediateQueueSize":"-2" ``` No job exception accompanies the negative metric. ### Zeta Engine Log No corresponding execution error was logged. The job remained healthy throughout the negative samples. ### Possible cause `IntermediateBlockingQueue.received()` makes the record visible in the blocking queue before incrementing the metric counter: ```java handleRecord(record, getIntermediateQueue()::put); intermediateQueueSize.inc(); ``` The consumer decrements the counter after polling the record: ```java Record<?> record = getIntermediateQueue().poll(100, TimeUnit.MILLISECONDS); if (record != null) { handleRecord(record, collector::collect); intermediateQueueSize.dec(); } ``` A consumer can poll and decrement the counter after `put()` but before the producer calls `inc()`, temporarily producing `-1` or `-2`. The fix should keep the metric update consistent with queue publication while also handling interrupted/failed `put()` and records dropped by the prepare-close path. ### Expected behavior `IntermediateQueueSize` should never be less than zero. ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's Code of 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]
