Joy-2000 opened a new issue, #19215:
URL: https://github.com/apache/hudi/issues/19215
### Describe the problem you faced
On the Flink streaming write path (`StreamWriteOperatorCoordinator` + the
write functions), a **same-graph global failover** (e.g. the JM commit fails in
`notifyCheckpointComplete` → `context.failJob()`, or
`jobmanager.execution.failover-strategy: full`) can silently **lose the write
metadata of an in-flight instant**, orphaning the already-written data/log
files so the instant never commits.
The root cause is that the coordinator can serialize a buffer entry for
`key=N` whose slot is **still all-null** (the instant was requested, but the
`WriteMetadataEvent` hasn't arrived yet). This all-null window exists because
**the instant is always requested before the data is flushed / the write events
are sent**:
- **`StreamWriteFunction`** — an **eager flush** (`bucket.isFull()` / memory
pool full, small `write.batch.size`) calls `instantToWrite()` inside
`flushBucket()`, i.e. **before** the checkpoint, then sends the event
afterwards.
- **`AppendWriteFunction`** — `initWriterHelper()` calls
`instantToWrite(true)` on the **first `processElement`** (lazily, as soon as
records start arriving). So `key=N` is created in the coordinator early in the
round, while `flushData()` only sends the `WriteMetadataEvent` later in
`snapshotState()` / `endInput()`. The all-null window is even wider here.
### To Reproduce
**Job setup**
- COW table, streaming write with `write.operation = insert` (so the writer
is `AppendWriteFunction`).
- No special config needed to open the all-null window:
`AppendWriteFunction` requests the instant on the **first record** via
`initWriterHelper()` → `instantToWrite(true)`, while the `WriteMetadataEvent`
is only sent later in `flushData()` during `snapshotState()` / `endInput()`. So
`key=N` exists in the coordinator (slot still all-null) well before the write
event arrives.
- Enable checkpointing so the coordinator serializes state each round.
**Steps**
1. Start the job and send at least one record so `key=N`'s instant is
requested (`initWriterHelper`) and the coordinator creates its (all-null)
buffer entry.
2. Let checkpoint `N+1` run: the all-null `key=N` entry is serialized into
the coordinator checkpoint state (this is the bug — see the trigger table
below).
3. Let the write event(s) for `key=N` arrive so the coordinator's live
buffer is fully populated.
4. Trigger a **same-graph global failover** — e.g. force a commit failure in
`notifyCheckpointComplete` (transient storage/lock error → `failJob()`), or set
`jobmanager.execution.failover-strategy: full` and kill a write task. Do
**not** stop/restart the whole job (that would be a *new-graph* failover,
`attemptId=0`, which recovers correctly).
5. After recovery, `key=N` is missing from the coordinator's buffer; on the
next checkpoint completion the instant commits empty. The data/log files
written for `key=N` are orphaned and never referenced → **silent data loss /
stuck inflight**.
**Whether `key=N` enters the checkpoint state (the trigger condition)**
| slot state at serialize time | passes `allMatch(e==null \|\|
e.isLastBatch())`? | serialized into state? | `putAll` overwrites live buffer?
| result |
|---|---|---|---|---|
| **all-null** — instant requested early (append first-record), no event yet
| **yes** (all elements are `null`) | **yes** | **yes** | ❌ data loss |
| **partial with an `isLastBatch=false` event** — an eager/intermediate
event already landed in the slot | **no** (that event is not `null` and
`isLastBatch=false`) | no | nothing to overwrite | ✅ safe (excluded by the
filter) |
| **key absent** — instant only requested in `snapshotState`, after
serialize | n/a | no | nothing to overwrite | ✅ safe |
The dangerous window is narrow: `key=N` must be **created but still entirely
null** at the exact moment `checkpointCoordinator` serializes. If even one
`isLastBatch=false` event has arrived, the entry is (correctly) excluded and
never enters the state; if no instant was requested yet, there is no entry at
all. Only the all-null-but-present case slips through the filter and later gets
overwritten by `putAll`.
**Timeline (the losing interleaving)**
```
write task coordinator
checkpoint state
---------- -----------
----------------
requestInstantTime(N) ────────────────────► initNewEventBuffer(key=N)
key=N slot = [null, null]
│
checkpointCoordinator(N+1) serializes
──────────────────────────────► { key=N: [null, null] } ← BUG: all-null
passes the filter
│
send WriteMetadataEvent(key=N) ────────────► live key=N = [event, event]
(fully populated)
│
JM commit fails → failJob()
│
resetToCheckpoint(N+1) → putAll
◄───────────────────────────── { key=N: [null, null] }
live key=N OVERWRITTEN back to
[null, null]
│
write task restart (attemptId>0)
sendBootstrapEvent → emptyBootstrap only (NO resend)
│
next notifyCheckpointComplete → commit key=N sees
all-null
→ files orphaned, instant
never completes ❌
```
### Expected behavior
We'd like feedback on whether the direction below is acceptable before
opening a PR.
**Fix 1 — never checkpoint an in-progress buffer**
(`EventBuffers.getAllCompletedEvents`): change the filter from "no eager-flush
event present" to "**every subtask reported its last batch**", reusing the
existing `allEventsReceived()`:
```java
// before
.filter(entry -> Arrays.stream(entry.getValue().getRight())
.allMatch(event -> event == null || event.isLastBatch()))
// after
.filter(entry -> allEventsReceived(entry.getValue().getRight()))
```
This excludes all-null / partially-filled entries from the coordinator
checkpoint state, so `resetToCheckpoint`'s `putAll` has nothing to overwrite
the live buffer with.
**Fix 2 — don't stop the heartbeat on commit failure**
(`HoodieFlinkWriteClient.releaseResources`): override to a no-op. The success
path already stops the heartbeat in `postCommit`; on failure the instant stays
INFLIGHT for retry after the coordinator's global failover, so the heartbeat
must survive to pass `abortIfHeartbeatExpired` on recommit. A genuinely
abandoned instant's heartbeat file goes stale and is reaped by the lazy clean
policy.
```java
@Override
public void releaseResources(String instantTime) {
// no-op: keep the heartbeat alive so an inflight instant can be
recommitted after a global failover
}
```
### Environment Description
- Hudi: 1.1.1
- Flink: 1.20.1
- Table type: MOR and COW; affects both `StreamWriteFunction` (MOR/COW
upsert) and `AppendWriteFunction` (append/insert)
### Additional context
_No response_
### Stacktrace
```shell
```
--
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]
