fhan688 commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3830283550
##########
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:
> 🤖 For a diverged bucket, reclamation only flushes that bucket, whereas the
creation-failed path flushes the largest bucket. If the diverged bucket is
small while other buckets hold most of the pool, the single retry can still
fail and throw "write buffer is too small to hold a single record" — a case the
old code would have gotten past (by flushing the largest bucket) at the cost of
corruption. Since disposing the diverged bucket is mandatory, would it be worth
also reclaiming the largest bucket as a fallback before giving up, so we don't
turn a previously-succeeding pipeline into a hard failure? @danny0405 does the
deferred pool-preemption follow-up cover this, or should the retry fall back to
largest-bucket reclamation here?
>
> ⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.
Good catch. Reclaiming only the diverged bucket may not release enough pages
when most of the memory is held by other buckets. I’ll update the write-failure
path to first flush and dispose the largest non-empty other bucket, when
present, and then flush and dispose the diverged bucket before retrying with a
fresh 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]
