[BUG][Flink] Data loss on Flink job restart: two StreamWriteOperatorCoordinators coexist while old close() is still running

### Bug Description

**What happened:**
When a user stopped a Flink job (not a cluster restart) and the job had already reached the CANCELLED terminal state, they immediately started a new job. The new job started up normally, but the next day partial data loss was discovered.

Root cause investigation: when the job entered the CANCELLED state, StreamWriteOperatorCoordinator.close() had not yet finished executing (HDFS was experiencing slow requests at the time, and close() took more than 10 minutes to complete). Before the old close() returned, the new job had already been launched and had created a new StreamWriteOperatorCoordinator. This resulted in two StreamWriteOperatorCoordinator instances coexisting and both committing instants to the same Hudi table. The instant/checkpoint states of the two coordinators overwrote or conflicted with each other, ultimately causing some already-committed data to be dropped.

Key timeline:
```shell
t0: User stops the job
t1: Job state on the JobManager -> CANCELLED (terminal)
t2: User starts a new job -> new Coordinator #2 is created and starts working
...
t1 + 10min: close() of the old Coordinator #1 finally finishes (HDFS slow requests)
-> #1 and #2 have coexisted for ~10 minutes, and both have been
writing/committing to the same table during that window
```

**What you expected:**
A normal stop → start cycle should not cause data loss. Any of the following would be acceptable:
1. When a new job starts, it should detect that the old StreamWriteOperatorCoordinator has not finished close() yet, and block/delay startup until the old coordinator has fully released its resources.
2. StreamWriteOperatorCoordinator should enforce mutual exclusion / leasing (e.g., a distributed lock or fencing token keyed on the table path) to guarantee that at any moment only one coordinator commits instants for a given table. A late-arriving coordinator should detect the conflict and fail fast; a coordinator that has lost its lease should stop committing.
3. The job state should not be allowed to transition to a terminal state (CANCELLED) until close() has fully returned; the job must not reach a terminal state while the coordinator is still doing asynchronous cleanup in the background.
4. Optimize the runtime of StreamWriteOperatorCoordinator.close(). Conceptually, close() should be a lightweight teardown operation — it should not perform heavy or long-blocking work (e.g., synchronous HDFS calls that can take 10+ minutes on a slow NameNode). Heavy work should either be moved out of the shutdown path, made asynchronous with a bounded timeout, or split so that the critical section that must run under shutdown is minimal. This both reduces the window in which two coordinators can coexist and prevents close() from being blocked indefinitely by external storage issues.

**Root cause of the data loss**

The data loss is caused by the following interleaving between the new coordinator and the still-alive old coordinator:

1. 22:31:38 — The new coordinator commits instant 20260721223123851 and then enters its scheduling phase. It does not immediately generate the next instant.
2. 22:31:43 — During this scheduling gap, the old coordinator (whose close() is still running due to slow HDFS) generates its own instant 20260721223143462.
3. Because the TaskManager fetches whatever instant is currently available via lastPending, it picks up the old coordinator's instant 20260721223143462 and starts flushing data against it. The corresponding WriteResultEvents are sent back and buffered on the new coordinator.
4. 22:34 — The new coordinator finally generates its next instant 20260721223443220. From this point on, the TaskManager flushes new data against 20260721223443220 and sends the resulting WriteResultEvents to the new coordinator.
5. When the new coordinator receives these WriteResultEvents from the TaskManager, it directly overwrites the events already in its own event buffer for that subtask. The writes that had been flushed against the old coordinator's instant 20260721223143462 are therefore silently dropped from the new coordinator's buffer, never make it into any commit metadata, and end up as lost data.

In short: within the new coordinator's scheduling gap, the TaskManager was tricked into flushing to an instant owned by the old coordinator; the resulting WriteResultEvents were then clobbered when the new coordinator's own instant came online, causing that batch of writes to be lost.

### Environment

**Hudi version:** 0.13.1
**Query engine:** Flink1.16
**Relevant configs:** mor-avro

### Logs and Stack Trace

The following two marker-file log lines come from the same writer subtask (Sink: bucket_write(t: (15/40)#0) within the same Flink checkpoint window, but they correspond to two different Hudi instants:
```shell
2026-07-21 22:32:27.316 WARN [Sink: bucket_write(t: (15/40)#0]
org.apache.hudi.table.marker.DirectWriteMarkers - Marker Path=
hdfs://xxxhudi/.hoodie/.temp/20260721223143462/dt=2026-07-21/
00000007-51bc-4193-8f56-3c9331d97cf4_14-40-0_20260721121340693.parquet.marker.APPEND
already exists, cancel creation

2026-07-21 22:36:29.215 INFO [Sink: bucket_write(t: (15/40)#0]
org.apache.hudi.table.marker.DirectWriteMarkers - [direct] Created marker file
hdfs://xxxhudi/.hoodie/.temp/20260721223443220/dt=2026-07-21/
00000007-51bc-4193-8f56-3c9331d97cf4_14-40-0_20260721121340693.parquet.marker.APPEND
in 213 ms
```

Start with StreamWriteOperatorCoordinator lifecycle and close(), then trace how TaskManager obtains the current instant and how WriteResultEvents are buffered. Review the DirectWriteMarkers logs and the stop→CANCELLED→restart timeline to reproduce the coexistence window. Done means a normal stop and restart cannot cause writes or commit state from one coordinator to be overwritten by another.

Written by the indexing model from the issue text.