[Bug] [Zeta] Batch job occasionally hangs in RUNNING forever under high-frequency job submission

### 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

In production we hit an intermittent problem: a **BATCH** job (JDBC MySQL source → JDBC
`MultiTableSink`) finished reading all source data and the bounded reader closed normally, but the
whole job then **stays in `RUNNING` forever and never turns `FINISHED`**. It cannot finish on its
own — the only way to clear it is a manual cancel or a master restart.

This does not happen every time; it shows up intermittently under a high job-submission load.
Below are the clues we collected. We'd like the community's help to analyze the possible causes and
directions for improvement.

Environment & workload characteristics

- SeaTunnel version: 2.3.13, Zeta engine.
- Deployment: separated master/worker, 2 masters (HA 32C+48G heap) + 8 workers(3 32C+48G heap + 5 16C+20G heap).
- **Workload**: the platform has **1000+ data-collection jobs**, submitted periodically by a
scheduler, with **on average several dozen job submissions per minute** (higher at peak). So the
cluster runs continuously under "sustained high-frequency submission + many small concurrent jobs".
- The job that hung is itself trivial: a single bounded table read, BATCH mode, and **no
`checkpoint.interval` configured (i.e. checkpoint disabled)**.

### SeaTunnel Version

SeaTunnel version: 2.3.13, Zeta engine

### SeaTunnel Config

```conf
for master：

-- seatunnel.yaml
seatunnel:
engine:
classloader-cache-mode: true
history-job-expire-minutes: 30
backup-count: 0
queue-type: blockingqueue
job-metrics-backup-interval: 60
print-execution-info-interval: 60
print-job-metrics-info-interval: 1800
job-metrics-partition-count: 4
job-schedule-strategy: WAIT
slot-service:
dynamic-slot: false
checkpoint:
interval: 60000
timeout: 600000
storage:
...
telemetry:
metric:
enabled: false
logs:
scheduled-deletion-enable: false
http:
enable-http: true
port: 8010
enable-dynamic-port: false

-- hazelcast-master.yaml
hazelcast:
cluster-name: platform
network:
rest-api:
enabled: false
endpoint-groups:
CLUSTER_WRITE:
enabled: true
DATA:
enabled: true
join:
...
port:
auto-increment: false
port: 5801
properties:
hazelcast.invocation.max.retry.count: 20
hazelcast.tcp.join.port.try.count: 30
hazelcast.logging.type: log4j2
hazelcast.operation.generic.thread.count: 50
hazelcast.operation.call.timeout.millis: 60000
hazelcast.operation.backup.timeout.millis: 60000
hazelcast.heartbeat.failuredetector.type: phi-accrual
hazelcast.heartbeat.interval.seconds: 5
hazelcast.max.no.heartbeat.seconds: 300
hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 10
hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200
hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 100
hazelcast.operation.thread.count: 32
hazelcast.io.thread.count: 32
member-attributes:
group:
type: string
value: default
team:
type: string
value: default

for worker:

-- seatunnel.yaml
seatunnel:
engine:
classloader-cache-mode: true
history-job-expire-minutes: 30
backup-count: 0
queue-type: blockingqueue
job-metrics-backup-interval: 60
print-execution-info-interval: 60
print-job-metrics-info-interval: 1800
job-metrics-partition-count: 4
job-schedule-strategy: WAIT
slot-service:
dynamic-slot: false
slot-num: 96
slot-allocate-strategy: slot_ratio
checkpoint:
interval: 60000
timeout: 600000
storage:
...
telemetry:
metric:
enabled: false
logs:
scheduled-deletion-enable: false
http:
enable-http: true
port: 8010
enable-dynamic-port: false

-- hazelcast-worker.yaml
hazelcast:
cluster-name: platform
network:
join:
...
port:
auto-increment: false
port: 5802
properties:
hazelcast.invocation.max.retry.count: 20
hazelcast.tcp.join.port.try.count: 30
hazelcast.logging.type: log4j2
hazelcast.io.thread.count: 32
hazelcast.operation.generic.thread.count: 50
hazelcast.operation.call.timeout.millis: 60000
hazelcast.operation.backup.timeout.millis: 60000
hazelcast.heartbeat.failuredetector.type: phi-accrual
hazelcast.heartbeat.interval.seconds: 5
hazelcast.max.no.heartbeat.seconds: 300
hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 10
hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200
hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 100
hazelcast.operation.thread.count: 32
member-attributes:
group:
type: string
value: default
team:
type: string
value: default
```

### Running Command

```shell
start up: ./bin/seatunnel-cluster.sh -d -r master ./bin/seatunnel-cluster.sh -d -r worker
submit job: /submit-job API use the JSON config
```

### Error Exception

```log
## Clues we found

### Clue 1: state of the three task threads while hung (from jstack)

| Role | Thread-stack location | Meaning |
|---|---|---|
| Source reader (taskGroup 2) | looping in `SourceFlowLifeCycle.collect:194` (sleep) | reader's internal `prepareClose=true`, but the **task-level `prepareCloseStatus=false`, still in RUNNING** |
| Sink side (taskGroup 2) | blocked in `IntermediateBlockingQueue.collect:54` (poll) | waiting forever for data / a close marker on the intermediate queue that never arrives |
| SplitEnumerator (taskGroup 1) | `SourceSplitEnumeratorTask.stateProcess:350` (PREPARE_CLOSE) | already detected all readers finished, already sent `LastCheckpointNotifyOperation` to master, waiting for `closeCalled` |

The source finished reading at `17:48:42`; after that there is **almost no further log** for this
job on either worker or master.

### Clue 2: how a bounded BATCH job is supposed to finish (for reference)

Normally a bounded BATCH job is closed by a "final checkpoint":

1. reader finishes → `signalNoMoreElement()` → notifies the enumerator;
2. enumerator detects all readers finished → sends `LastCheckpointNotifyOperation` to master →
`CheckpointCoordinator.readyToClose()`;
3. master triggers the final checkpoint `tryTriggerPendingCheckpoint(COMPLETED_POINT_TYPE)`;
4. this close barrier flows enumerator → reader → sink, each acks, and it sets `prepareCloseStatus`
to true on every task, advancing each task `RUNNING → PREPARE_CLOSE → CLOSED`;
5. once all complete, master tells the tasks to close and the job finishes.

In the hung job, step 3 appears to have happened (the enumerator is already in PREPARE_CLOSE), but
the **source reader's `prepareCloseStatus` was never set** — i.e. the close-barrier handshake never
completed / never reached the reader.

### Clue 3: in the same time window, worker↔master Hazelcast communication was massively timing out and the connection was flapping

In the logs of the worker that ran the source/sink (172.18.1.145) and the active master
(172.18.1.143), **right in the window when the job tried to close (around source close at
`17:48:42`)**, cluster-internal communication was badly degraded:

- the master logged **1419 `Slow operation detected: ReportMetricsOperation`** warnings within one
hour; `ReportMetricsOperation` (partitionId=-1) was hogging almost all generic-operation threads;
- several operations from the worker to the master were **rejected by the 60s
operation-call-timeout**, e.g. `JobEventReportOperation` (17:46:36, 17:48:03, with the connection
already `alive=false`) and `ReportMetricsOperation` (17:48:59, `ERROR ... failed to update metrics`);
- the TCP cluster connection worker(145)↔master(143) was **re-initialized 6 times between 17:39 and
17:56** (each reconnect drops the operations in flight on that connection); meanwhile **no
Hazelcast cluster membership change occurred** (no master failover) — only the connection flapped;
- shortly after, `SourceRegisterOperation` was flagged as a slow operation (17:50:37) and then
**failed after 30 retries** (17:51:27), confirming operations were being lost during this period.

### Clue 4: the close handshake relies on a chain of Hazelcast operations

The close flow in step 2 is, in implementation, a chain of cross-node Hazelcast operations:
`LastCheckpointNotifyOperation` (worker→master), `CheckpointBarrierTriggerOperation`
(master→enumerator), `BarrierFlowOperation` (enumerator→reader), `TaskAcknowledgeOperation`
(reader→master). If any one of these is lost (in a connection drop / 60s-timeout rejection) and is
not retried, the close flow cannot complete.

We also noticed that when `checkpoint.interval` is not configured (checkpoint disabled), this final
checkpoint used to close the bounded source does not seem to have the same timeout detection that a
normal checkpoint has (this is only an observation; we're not sure whether it is by design).

## Our questions / what we'd like the community to look at

1. Under our "1000+ jobs, several dozen submissions per minute" high-frequency scenario,
`ReportMetricsOperation` and similar internal operations saturate the master's operation threads
and cause the worker↔master connection to repeatedly time out / reconnect — is this a known
scalability bottleneck? How does the community usually mitigate it?
2. For this phenomenon, what does the community think are the **possible root causes** and the
**directions for future optimization**? (Whether at the config, deployment, or engine level —
we'd love to hear any thoughts.)
```

### Zeta or Flink or Spark Version

_No response_

### Java or Scala Version

Java: 8

### Screenshots

_No response_

### 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)

Start by tracing the close-handshake entry points named in the report: SourceFlowLifeCycle.collect:194, IntermediateBlockingQueue.collect:54, SourceSplitEnumeratorTask.stateProcess:350, and the listed Hazelcast operations. Reproduce the hang under high-frequency submission while collecting worker/master logs and thread dumps; the issue does not define a concrete fix or acceptance test, so completion would require agreeing on the root cause and expected recovery behavior.

Written by the indexing model from the issue text.