fhan688 opened a new pull request, #19692:
URL: https://github.com/apache/hudi/pull/19692
### Describe the issue this Pull Request addresses
Part of #19664.
When the shared Flink write-memory pool is exhausted,
`BinaryInMemorySortBuffer.write(RowData)` may return `false` after its
variable-length storage has already been partially mutated, even though no
index entry was committed for the failed row.
`StreamWriteFunction` previously handled this case in the same way as a
failure to create a new buffer. It could flush another bucket and then retry
the record against the same `RowDataBucket`. Reusing such a diverged buffer may
resolve variable-length offsets against inconsistent memory segments, leading
to corrupted buffered values or runtime exceptions.
This is the correctness-focused PR1 proposed in #19664. Owner-aware shared
memory-pool preemption remains outside the scope of this PR and can be
introduced separately.
### Summary and Changelog
This PR prevents a write buffer from being reused after
`BinaryInMemorySortBuffer.write` returns `false`, and makes the corresponding
memory-exhaustion recovery path exception-safe.
Changes include:
- Introduce explicit buffer-write outcomes to distinguish:
- failure to create a new buffer because no memory pages are available;
- failure while writing into an existing buffer, leaving that buffer
potentially diverged.
- Mark a `RowDataBucket` as terminally diverged when its underlying buffer
returns `false`.
- Reject all subsequent writes to a diverged bucket.
- Recover a diverged non-empty bucket by flushing only the successfully
indexed records and then disposing the bucket.
- Skip flushing a diverged empty bucket and dispose it directly.
- Preserve the existing largest-bucket reclamation strategy only for
buffer creation failures.
- Retry the failed record once using a newly created buffer.
- Ensure pages are returned when recovery flushing or retrying fails.
- Dispose remaining buckets during checkpoint flushing and operator
shutdown, including empty buckets.
- Preserve cleanup failures as suppressed exceptions during operator close.
Tests added:
- A focused `RowDataBucket` test verifying that:
- memory exhaustion causes the bucket to become diverged;
- a second write is rejected;
- previously successful records and variable-length values remain
readable;
- disposing the bucket returns all allocated memory pages.
- An end-to-end Flink bucket-write test using:
- a 1 MB shared write-memory pool;
- four buckets;
- forty records with 256 KB variable-length payloads and map values.
- The end-to-end test verifies that:
- memory-exhaustion recovery flushing is triggered;
- no records are lost or duplicated;
- variable-length string and map values remain intact;
- all memory pages are returned after checkpoint flushing.
Verification result:
```text
Tests run: 2, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS
```
No code was copied from another project.
### Impact
There are no public API, configuration, or storage-format changes.
For normal writes where the memory pool is not exhausted, buffering
behavior is unchanged.
Under memory pressure, Hudi now discards a buffer that may have diverged
instead of attempting to reuse it. This prevents data corruption and runtime
failures. It may cause an additional eager flush only on the memory-exhaustion
recovery path.
The change also improves memory-page cleanup on exceptional flush, retry,
checkpoint, and shutdown paths.
### Risk Level
medium
This changes the Flink write path used during memory exhaustion and
therefore touches data-correctness-sensitive recovery behavior.
The risk is mitigated by:
- keeping the normal successful buffering path unchanged;
- distinguishing buffer creation failures from existing-buffer write
failures;
- retrying only once on a fresh buffer;
- flushing only successfully indexed records from a diverged buffer;
- adding focused buffer-level coverage;
- adding end-to-end multi-bucket coverage that validates record count,
variable-length values, recovery flushing, and complete page restoration.
### Documentation Update
none
This is an internal correctness fix with no new configuration, public API,
or user-facing workflow.
### Contributor's checklist
- [x] Read through contributor's guide
(https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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]
