fhan688 commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3828784807
##########
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:
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]
