hudi-agent commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3822845835
##########
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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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;
+ }
+
+ ValidationUtils.checkState(
Review Comment:
🤖 nit: the message "Unexpected successful buffer write result during memory
reclamation" is misleading — `SUCCESS` is already excluded by the caller's `if
(result != SUCCESS)` guard, so this assertion would only fire for a future enum
constant, not a `SUCCESS` result. Could you use something like `"Unexpected
BufferWriteResult in reclaimMemoryAfterFailedWrite: " + result` so a future
debugger isn't sent chasing how SUCCESS got here?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
