cshuo commented on code in PR #19692:
URL: https://github.com/apache/hudi/pull/19692#discussion_r3840431975
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java:
##########
@@ -315,19 +314,22 @@ private String getBucketID(String partitionPath, String
fileId) {
* <p>2. Data Bucket exists, but fails to request new memory pages from
memory pool.
*/
private boolean doBufferRecord(String bucketID, HoodieFlinkInternalRow
record) throws IOException {
- try {
- RowDataBucket bucket = this.buckets.computeIfAbsent(bucketID,
- k -> new RowDataBucket(
- bucketID,
- createDataBuffer(),
- getBucketInfo(record),
- this.config.get(FlinkOptions.WRITE_BATCH_SIZE)));
-
- return bucket.writeRow(record.getRowData());
- } catch (MemoryPagesExhaustedException e) {
- log.info("There is no enough free pages in memory pool to create buffer,
need flushing first.", e);
- return false;
+ RowDataBucket bucket = this.buckets.get(bucketID);
Review Comment:
This change is unnecessary now?
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java:
##########
@@ -382,8 +371,107 @@ 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);
+ if (failedBucket != null) {
Review Comment:
There are two similar code to get the largest data bucket here, can we
simplify this by selecting the largest non-empty bucket once, excluding the
current `bucketID`?
BTW, there are several new `*dispose*` helpers, making the flow a bit harder
to follow. Could you have a another little refactoring to make the flow
cleaner? Such as merging `reclaimMemoryAfterFailedWrite` and
`flushAndDisposeAfterWriteFailure` into one.
--
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]
