gkhnelbstn opened a new issue, #12411: URL: https://github.com/apache/seatunnel/issues/12411
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Bug%22) and found no similar issues. ### What happened An `Http` source with `poll_interval_millis` set — that is, a streaming job — never completes a checkpoint, and Zeta fails the job at the checkpoint timeout. The reader is demonstrably still working while this happens: `TableSourceReceivedCount` grew by four rows every 25 s (two records, a ten-second poll) right up to the failure. The cause is that the wait between polls is done while holding the checkpoint lock. `HttpSourceReader.pollNext` takes it around the whole poll: ```java @Override public void pollNext(Collector<SeaTunnelRow> output) throws Exception { synchronized (output.getCheckpointLock()) { internalPollNext(output); } } @Override public void internalPollNext(Collector<SeaTunnelRow> output) throws Exception { try { ... } finally { if (Boundedness.BOUNDED.equals(context.getBoundedness()) && noMoreElementFlag) { context.signalNoMoreElement(); } else { if (httpParameter.getPollIntervalMillis() > 0) { Thread.sleep(httpParameter.getPollIntervalMillis()); // lock still held } } } } ``` That is the same lock `SourceFlowLifeCycle#triggerBarrier` needs, as the engine's own comment in `SourceFlowLifeCycle` says: > The current thread obtain a checkpoint lock in the method > `SourceReader#pollNext(Collector)`. When trigger the checkpoint or savepoint, > other threads try to obtain the lock in the method > `SourceFlowLifeCycle#triggerBarrier(Barrier)`. When high CPU load, checkpoint > process may be blocked as long time. So we need sleep to free the CPU. The engine adds a `Thread.sleep(0L)` between polls for exactly this reason, because the window between one `pollNext` releasing the lock and the next taking it is narrow. A poll interval turns that into one narrow window every `poll_interval_millis`, and `synchronized` is not fair, so the barrier thread can lose the race indefinitely. Raising the checkpoint interval or the timeout does not help — a contended lock is not a slow operation. Tried, with a ten-second poll: | `checkpoint.interval` | `checkpoint.timeout` | result | |---|---|---| | 5 000 ms | 60 000 (server default) | FAILED after ~2 min | | 20 000 ms | 60 000 (server default) | FAILED after ~2 min | | 20 000 ms | 60 000 stated in the job env | FAILED after ~2 min | ### Suggested fix `Object.wait(long)` releases the monitor while it waits; `Thread.sleep` does not. The monitor is held at that point, so this needs no other change, and a spurious wake-up only means polling slightly early: ```diff - Thread.sleep(httpParameter.getPollIntervalMillis()); + output.getCheckpointLock().wait(httpParameter.getPollIntervalMillis()); ``` With that applied the same job checkpoints normally and keeps delivering every poll, instead of failing at about two minutes. ### SeaTunnel Version 2.3.13, and `dev` has the same code at the time of writing. ### SeaTunnel Config ```conf env { job.mode = "STREAMING" parallelism = 1 checkpoint.interval = 20000 } source { Http { plugin_output = "src" url = "http://example.internal:8099/members"; method = "GET" format = "json" content_field = "$.members.*" poll_interval_millis = 10000 schema { fields { member_no = "bigint", full_name = "string" } } } } sink { Jdbc { ... } } ``` ### Running Command ```shell curl -X POST "http://localhost:8080/submit-job?jobName=poll_a_listing"; \ -H 'Content-Type: application/json' -d @job.json ``` ### Error Exception ```log org.apache.seatunnel.engine.server.checkpoint.CheckpointException: Checkpoint expired before completing. Please increase checkpoint timeout in the seatunnel.yaml or jobConfig env. at org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinator.handleCoordinatorError(CheckpointCoordinator.java:289) at org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinator.lambda$null$9(CheckpointCoordinator.java:713) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750) ``` ### Zeta or Flink or Spark Version Zeta, the cluster from the official 2.3.13 image. ### Java or Scala Version Java 8 (`1.8.0_502`), the version in the official image. ### Screenshots _No response_ ### 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](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]
