hudi-agent commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3845826933
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java:
##########
@@ -382,8 +368,103 @@ 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());
+ }
+ }
Review Comment:
🤖 nit: the new private methods use `bucketID` (all-caps ID) while the
existing getter is `getBucketId()` — could you align to `bucketId` across
`reclaimMemoryAfterFailedWrite`, `retryBufferRecord`, and
`disposeFailedRetryBucket` to stay consistent with the rest of the class?
<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 +368,103 @@ 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) {
+ // A creation failure leaves no bucket in the map, while a write failure
leaves the
+ // diverged bucket in the map so that its committed records can be flushed
and disposed.
+ RowDataBucket failedBucket = this.buckets.get(bucketID);
+ RowDataBucket bucketToFlush = this.buckets.values().stream()
+ .filter(bucket -> !bucketID.equals(bucket.getBucketId()) &&
!bucket.isEmpty())
+ .max(Comparator.comparingLong(RowDataBucket::getBufferSize))
+ .orElse(null);
+
+ if (failedBucket == null) {
+ if (bucketToFlush == null) {
+ throw new HoodieException(
+ "Not enough memory pages to create a RowData buffer and no
non-empty bucket can be flushed");
+ }
+ flushAndDisposeBucket(bucketToFlush);
+ return;
+ }
+
+ ValidationUtils.checkState(
+ failedBucket.isDiverged(), "The failed RowData bucket has not
diverged");
+
+ RuntimeException failure = null;
+ if (bucketToFlush != null) {
+ try {
+ flushAndDisposeBucket(bucketToFlush);
+ } catch (RuntimeException e) {
+ failure = e;
+ }
+ }
+
+ try {
+ flushAndDisposeBucket(failedBucket);
+ } catch (RuntimeException e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private void retryBufferRecord(
+ String bucketID, HoodieFlinkInternalRow record) throws IOException {
+ final boolean success;
+ try {
+ success = doBufferRecord(bucketID, record);
+ } catch (IOException | RuntimeException e) {
+ disposeFailedRetryBucket(bucketID, e);
+ throw e;
+ }
+
+ if (!success) {
+ HoodieException exception = new HoodieException(
+ this.buckets.get(bucketID) == null
+ ? "Not enough memory pages to create a RowData buffer after
flushing"
+ : "The write buffer is too small to hold a single record");
+ disposeFailedRetryBucket(bucketID, exception);
+ throw exception;
+ }
+ }
+
+ private void disposeFailedRetryBucket(String bucketID, Throwable failure) {
+ RowDataBucket bucket = this.buckets.get(bucketID);
+ if (bucket == null) {
+ return;
+ }
+ try {
+ disposeBucket(bucket);
+ } catch (RuntimeException cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+
+ private void flushAndDisposeBucket(RowDataBucket bucket) {
+ long bufferSize = bucket.getBufferSize();
+ try {
+ if (!bucket.isEmpty()) {
+ flushBucket(bucket);
+ }
Review Comment:
🤖 nit: the nested `try { countDown } finally { disposeBucket }` inside the
outer `finally` block is a bit surprising — if `tracer.countDown` is
non-throwing, a flat `finally { countDown; disposeBucket; }` would be clearer;
if it can throw unchecked, a short comment here explaining why would help the
next reader.
<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]
