fhan688 opened a new issue, #19664:
URL: https://github.com/apache/hudi/issues/19664
### Bug Description
**What happened:**
The Flink stream writer may reuse a `BinaryInMemorySortBuffer` after
`write()` has returned `false` because the shared `MemorySegmentPool` was
exhausted.
When memory exhaustion happens in the middle of record serialization,
Flink's `BinaryInMemorySortBuffer.write()` catches an `EOFException` and
returns `false`. However, the underlying paged output position may already have
advanced, while the buffer's logical record offset is only updated after
successful serialization.
The current Hudi recovery path flushes the globally largest bucket and
then retries the record on its original bucket. If the failed bucket is not the
largest bucket, another bucket is flushed and the failed buffer is reused. This
may cause the sort-index pointer to no longer match the record's physical
location.
The corruption is usually detected later, when the bucket is flushed and
the buffered `RowData` is read back. Variable-length fields such as `STRING`,
`MAP`, and `ARRAY` make the problem easier to reproduce.
Relevant Hudi code:
- `StreamWriteFunction#doBufferRecord`:
https://github.com/apache/hudi/blob/42e885f9678321711e0d1563b1f651efac5794f5/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L317-L331
- `StreamWriteFunction#bufferRecord`:
https://github.com/apache/hudi/blob/42e885f9678321711e0d1563b1f651efac5794f5/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java#L344-L367
- `RowDataBucket#writeRow`:
https://github.com/apache/hudi/blob/42e885f9678321711e0d1563b1f651efac5794f5/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/RowDataBucket.java#L63-L69
- `HeapMemorySegmentPool#nextSegment`:
https://github.com/apache/hudi/blob/42e885f9678321711e0d1563b1f651efac5794f5/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/buffer/HeapMemorySegmentPool.java#L60-L69
The same underlying behavior is present in the supported Flink versions,
including Flink 1.18, 1.19, 1.20, 2.0, and 2.1:
`BinaryInMemorySortBuffer.write()` returns `false` after an `EOFException`
without rolling back partially written record bytes.
**What you expected:**
A failed `BinaryInMemorySortBuffer.write()` must never result in the same
buffer being reused for another write.
When the shared memory pool is exhausted, Hudi should either:
1. Flush another bucket and return its pages before the current
serialization fails; or
2. Treat the current buffer as invalid, flush only its previously
committed records, dispose it, and retry the record on a fresh buffer.
The writer must preserve all records and variable-length field values
without corrupting `RowData`.
**Steps to reproduce:**
A deterministic regression test can be constructed with the following setup:
1. Create a Flink MOR table using bucket index and `upsert`.
2. Configure four buckets so that one writer task maintains multiple
active `RowDataBucket` instances.
3. Use a small write memory pool, for example:
```java
conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 201.0);
conf.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 100);
```
This leaves approximately 1 MB for the write buffer.
4. Use a schema containing:
- a record key;
- a large variable-length `STRING`;
- a `MAP<STRING, STRING>`;
- an ordering timestamp;
- a partition field.
5. Write approximately 40 records, each containing a 256 KB string, with
keys distributed across four buckets.
6. Trigger a checkpoint so that all remaining buckets are read and flushed.
Example configuration:
```java
conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
conf.set(FlinkOptions.OPERATION, "upsert");
conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
conf.set(FlinkOptions.INDEX_TYPE, "BUCKET");
conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 4);
conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 201.0);
conf.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 100);
```
The important condition is that the bucket whose write fails is not the
globally largest bucket. In that case, the current implementation flushes
another bucket and retries the record on the failed buffer.
## Root cause
The failure sequence is:
```text
BinaryInMemorySortBuffer.write(record)
-> record serialization requests another MemorySegment
-> MemorySegmentPool.nextSegment() returns null
-> serialization throws EOFException
-> BinaryInMemorySortBuffer.write() returns false
-> paged output may already contain partial record bytes
-> Hudi flushes the globally largest bucket
-> if that is a different bucket, Hudi retries the failed buffer
-> logical sort-index offset and physical record location diverge
-> corrupted RowData is observed during a later flush
```
`RowDataBucket` currently has no state indicating that its underlying
buffer has diverged, so there is no protection against writing to it again.
The problem applies to both memory modes:
- `HeapMemorySegmentPool` returns `null` when all pages are used.
- Flink's `LazyMemorySegmentPool`, used by managed memory mode, also
returns `null` when all pages are used.
## Proposed approach
I propose implementing two layers of protection.
### 1. Correctness guard
- Mark a `RowDataBucket` as diverged whenever
`BinaryInMemorySortBuffer.write()` returns `false`.
- Reject every subsequent write to a diverged buffer.
- Distinguish between:
- failure to create a new buffer because fewer than the minimum pages
are available; and
- failure while writing to an existing buffer.
- For an existing diverged buffer:
- flush its previously committed records;
- dispose the buffer, including partial bytes from the failed record;
- retry the record once on a fresh buffer.
- If the retry also fails, dispose the failed buffer and fail the task
with a clear error.
- Fail fast if a non-empty bucket cannot be flushed because there is no
inflight instant.
- Ensure all pages are returned before closing a managed memory pool.
This layer independently prevents data corruption.
### 2. Owner-aware memory preemption
Add a Hudi-owned `MemorySegmentPool` wrapper that can decorate either:
- `HeapMemorySegmentPool`; or
- Flink's `LazyMemorySegmentPool`.
Before writing a record, the writer registers the current bucket as the
memory owner. When the delegate pool returns `null`, the wrapper asks the
writer to flush the largest non-empty bucket other than the current owner.
The pages returned by disposing that bucket are then used to continue the
current serialization without allowing it to fail halfway through.
The owner should be scoped and cleared in a `finally` block. The
preemption logic must never preempt the in-flight bucket and must not retry
recursively without a progress check.
## Proposed PR plan
### PR 1: Correctness fix
Suggested title:
```text
fix(flink): prevent reuse of diverged write buffers on memory exhaustion
```
Scope:
- Add the diverged state to `RowDataBucket`.
- Prevent writes to a buffer after `write()` returns `false`.
- Flush and dispose the failed current bucket before retrying.
- Retry only on a fresh buffer and only once.
- Add fail-fast and cleanup behavior.
- Add regression coverage for both on-heap and managed memory.
- Verify record count and variable-length field values, rather than only
checking that no exception is thrown.
### PR 2: Memory preemption optimization
Suggested title:
```text
perf(flink): preempt inactive write buckets on memory exhaustion
```
Scope:
- Add an owner-aware `MemorySegmentPool` wrapper.
- Support both heap and managed memory delegates.
- Never preempt the bucket currently being written.
- Flush the largest other bucket when memory pages are exhausted.
- Add unit tests for owner handling, page return, bounded retry, and
delegate closing.
- Add integration coverage for regular and LSM stream writes.
The first PR provides the correctness guarantee. The second PR prevents
most mid-record allocation failures and avoids unnecessary disposal and
reconstruction of the current bucket.
## Acceptance criteria
- A buffer is never written again after `BinaryInMemorySortBuffer.write()`
returns `false`.
- Multi-bucket writes under a constrained memory pool complete without
RowData corruption.
- Single-bucket writes recover by flushing committed records and
recreating the buffer.
- Both `ON_HEAP` and `MANAGED` memory modes are covered.
- Regular MOR/COW and LSM stream writes are covered.
- The number of output records matches the number of input records.
- Large `STRING`, `MAP`, and `ARRAY` values remain intact.
- Managed memory is fully returned when the writer closes.
- An oversized record fails deterministically without an infinite retry
loop.
- Existing checkpoint and exactly-once semantics remain unchanged.
### Environment
**Hudi version:** Apache Hudi master at
`42e885f9678321711e0d1563b1f651efac5794f5`
**Query engine:** Apache Flink
**Flink versions:** The behavior has been verified in the relevant
`BinaryInMemorySortBuffer` implementation for Flink 1.18.x, 1.19.x, 1.20.x,
2.0.x, and 2.1.x. Hudi master currently defaults to Flink 2.1.1.
**Table type:** MERGE_ON_READ
**Write operation:** UPSERT
**Index type:** BUCKET
**Relevant configs:**
```java
conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
conf.set(FlinkOptions.OPERATION, "upsert");
conf.set(FlinkOptions.RECORD_KEY_FIELD, "uuid");
conf.set(FlinkOptions.INDEX_TYPE, "BUCKET");
conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS, 4);
conf.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 201.0);
conf.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 100);
The issue is relevant to both:
- write.buffer.memory.type=ON_HEAP
- write.buffer.memory.type=MANAGED
### Logs and Stack Trace
**Logs and Stack Trace**
The exact exception depends on the corrupted variable-length field and
whether JVM assertions are enabled.
Representative failures observed while reading the corrupted RowData
during bucket flush include:
java.lang.ArrayIndexOutOfBoundsException
and:
java.lang.NegativeArraySizeException
With assertions enabled, Flink binary data classes may fail earlier with
an assertion error while resolving a corrupted MAP or ARRAY length.
The exception is a delayed symptom. The actual corruption occurs earlier
when BinaryInMemorySortBuffer.write() returns false after partial serialization
and Hudi later retries the same buffer.
--
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]
