fhan688 commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3830290733
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java:
##########
@@ -382,8 +379,79 @@ protected void bufferRecord(HoodieFlinkInternalRow record)
throws IOException {
* RowData data bucket can not be used after disposing.
*/
private void disposeBucket(RowDataBucket rowDataBucket) {
- rowDataBucket.dispose();
- this.buckets.remove(rowDataBucket.getBucketId());
+ try {
+ rowDataBucket.dispose();
+ } finally {
+ this.buckets.remove(rowDataBucket.getBucketId());
+ }
+ }
+
+ private void reclaimMemoryAfterFailedWrite(String bucketID,
BufferWriteResult result) {
+ if (result == BufferWriteResult.BUFFER_DIVERGED) {
+ RowDataBucket divergedBucket = this.buckets.get(bucketID);
+ ValidationUtils.checkState(
+ divergedBucket != null && divergedBucket.isDiverged(),
+ "The failed RowData bucket is missing or has not diverged");
+ flushAndDisposeBucket(divergedBucket);
+ return;
Review Comment:
> +1. For a diverged buffer, we should also reclaim the largest other bucket
so that retrying with a fresh buffer has sufficient memory.
>
> The recovery behavior should therefore be:
>
> * `BUFFER_DIVERGED`: (1) flush and dispose the largest non-empty bucket
other than the diverged bucket, when present; (2) flush the committed records
from the diverged bucket and dispose it; then retry the record on a fresh
buffer.
> * `BUFFER_CREATION_FAILED`: flush and dispose the largest non-empty
bucket, then retry buffer creation.
>
> Once the owner-aware memory pool is introduced, the `BUFFER_DIVERGED` path
can be optimized further: when the pool is about to be exhausted, it can flush
the largest non-owner bucket before the current serialization fails. This would
allow the current write to continue and avoid unnecessarily flushing and
recreating the current bucket. @fhan688
Thanks for the clarification. I’ll update the recovery behavior accordingly:
- For a buffer write failure, flush and dispose the largest non-empty
bucket other than the failed bucket when present, then flush and dispose the
failed bucket, and retry the record with a fresh buffer.
- For a buffer creation failure, flush and dispose the largest non-empty
bucket, then retry buffer creation.
The owner-aware preemption can remain a follow-up optimization.
--
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]
