fhan688 opened a new pull request, #19728:
URL: https://github.com/apache/hudi/pull/19728
### Describe the issue this Pull Request addresses
Closes #19664.
This is a follow-up to #19692, which made the Flink writer gracefully fall
back to a diverged-record bucket when a write bucket cannot acquire additional
memory segments.
However, the current bucket may fail to allocate a memory segment while
reclaimable segments are still retained by other inactive write buckets. In
that case, directly using the diverged-record fallback can cause unnecessary
bucket divergence and may leave reclaimable managed memory unused.
This PR allows the writer to preemptively flush and dispose an inactive
write bucket before falling back to the existing divergence path.
### Summary and Changelog
When a write bucket cannot allocate another memory segment, the writer now
attempts to reclaim memory from the largest non-empty inactive bucket and
retries the allocation once.
The changes include:
- Add `PreemptiveMemorySegmentPool`, which wraps an existing
`MemorySegmentPool`.
- Track the bucket currently performing `writeRow` as the active memory
owner.
- Trigger a memory-reclamation callback when segment allocation fails.
- Select the largest non-empty bucket other than the active bucket as the
reclamation victim.
- Flush and dispose the selected bucket using the lifecycle handling
introduced in #19692.
- Retry the failed segment allocation exactly once after successful
reclamation.
- Prevent recursive preemption while a reclamation callback is running.
- Preserve the existing diverged-record fallback when:
- there is no eligible inactive bucket;
- reclamation does not release enough memory; or
- the current record still cannot fit into the bounded memory pool.
- Reuse the inactive-bucket selection logic in the existing failed-write
reclamation path.
- Add focused unit tests for heap and Flink managed-memory pools.
- Extend the memory-exhaustion integration test to cover both the default
and LSM Tree storage layouts.
No code was copied from another project.
### Impact
There are no changes to the storage format, public APIs, configuration
options, or default configuration values.
For normal writes with sufficient memory, behavior remains unchanged apart
from lightweight active-bucket tracking.
Under memory pressure, an inactive bucket may be flushed earlier so that
its memory segments can be reused by the active bucket. This reduces
unnecessary bucket divergence and improves progress when memory is distributed
across multiple buffered buckets.
Earlier flushing under severe memory pressure may produce smaller write
batches, but this only occurs after a segment allocation has already failed.
### Risk Level
Medium.
The change affects the Flink writer's bucket flushing and managed-memory
lifecycle under memory exhaustion. The risk is mitigated by:
- retrying allocation only once;
- excluding the currently active bucket from reclamation;
- preventing recursive preemption;
- reusing the existing flush-and-dispose lifecycle;
- preserving the fallback behavior introduced in #19692;
- verifying that all managed-memory pages are returned;
- testing both the default and LSM Tree bucket implementations.
The following validations passed:
- `TestPreemptiveMemorySegmentPool`: 5 tests passed.
- `TestBucketStreamWriteMemoryExhaustion`: 2 parameterized scenarios
passed for the default and LSM Tree layouts.
- Flink 1.18 reactor compilation passed.
- Flink 2.1 adapter build passed.
- Checkstyle, RAT, and `git diff --check` passed.
### Documentation Update
None.
This is an internal memory-reclamation improvement and does not introduce
or modify any user-facing configuration or API.
### 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]
