KnightChess opened a new issue, #19357:
URL: https://github.com/apache/hudi/issues/19357
### 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**
<img width="2920" height="926" alt="Image"
src="https://github.com/user-attachments/assets/aad6b0af-67fe-48b9-b3b8-f5dcf62eb7d0";
/>
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 old 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:** Flink
**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
```
--
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]
