You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2025/12/06 08:44:14 UTC
(hudi) branch master updated: fix: Remove explicit casting to HoodieWriteMergeHandle in Flink commit action executor (#13590)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 5c817c3ccff8 fix: Remove explicit casting to HoodieWriteMergeHandle in Flink commit action executor (#13590)
5c817c3ccff8 is described below
commit 5c817c3ccff825d32d1a1a26826ac6a70d5959d2
Author: Shuo Cheng <nj...@gmail.com>
AuthorDate: Sat Dec 6 16:44:07 2025 +0800
fix: Remove explicit casting to HoodieWriteMergeHandle in Flink commit action executor (#13590)
---
.../org/apache/hudi/io/HoodieWriteMergeHandle.java | 4 ----
.../hudi/io/FlinkFileGroupReaderBasedMergeHandle.java | 6 ------
.../action/commit/BaseFlinkCommitActionExecutor.java | 18 +++++-------------
.../java/org/apache/hudi/sink/StreamWriteFunction.java | 10 +++++++++-
.../hudi/utils/factory/CollectSinkTableFactory.java | 4 ++--
5 files changed, 16 insertions(+), 26 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
index 8091ca4ac5fd..03c1af95e012 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
@@ -244,10 +244,6 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends HoodieAbstractMergeHandl
}
}
- public boolean isEmptyNewRecords() {
- return keyToNewRecords.isEmpty();
- }
-
protected boolean writeUpdateRecord(HoodieRecord<T> newRecord, HoodieRecord<T> oldRecord, HoodieRecord combineRecord, Schema writerSchema) throws IOException {
boolean isDelete = false;
if (oldRecord.getData() != combineRecord.getData()) {
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
index 9420752615f3..7ab517662b2a 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkFileGroupReaderBasedMergeHandle.java
@@ -118,12 +118,6 @@ public class FlinkFileGroupReaderBasedMergeHandle<T, I, K, O>
writeMarkers.createIfNotExists(partitionPath, dataFileName, getIOType());
}
- @Override
- public boolean isEmptyNewRecords() {
- // `keyToNewRecords` is not initialized or used for file group reader based handle.
- return false;
- }
-
@Override
boolean needsUpdateLocation() {
// No need to update location for Flink hoodie records because all the records are pre-tagged
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
index d713ead1d53a..7a4d9f90e5e9 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/BaseFlinkCommitActionExecutor.java
@@ -27,11 +27,13 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.execution.FlinkLazyInsertIterable;
import org.apache.hudi.io.ExplicitWriteHandleFactory;
import org.apache.hudi.io.HoodieCreateHandle;
+import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.io.HoodieWriteMergeHandle;
import org.apache.hudi.io.HoodieWriteHandle;
import org.apache.hudi.io.IOUtils;
@@ -43,7 +45,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.time.Duration;
-import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -190,24 +191,15 @@ public abstract class BaseFlinkCommitActionExecutor<T> extends
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr)
throws IOException {
- // This is needed since sometimes some buckets are never picked in getPartition() and end up with 0 records
- HoodieWriteMergeHandle<?, ?, ?, ?> upsertHandle = (HoodieWriteMergeHandle<?, ?, ?, ?>) this.writeHandle;
- if (upsertHandle.isEmptyNewRecords() && !recordItr.hasNext()) {
- LOG.info("Empty partition with fileId => {}.", fileId);
- return Collections.singletonList((List<WriteStatus>) Collections.EMPTY_LIST).iterator();
- }
+ ValidationUtils.checkArgument(this.writeHandle instanceof HoodieMergeHandle,
+ "`writeHandle` should be an instance of `HoodieMergeHandle`");
// these are updates
- return IOUtils.runMerge(upsertHandle, instantTime, fileId);
+ return IOUtils.runMerge((HoodieMergeHandle<?, ?, ?, ?>) this.writeHandle, instantTime, fileId);
}
@Override
public Iterator<List<WriteStatus>> handleInsert(String idPfx, Iterator<HoodieRecord<T>> recordItr)
throws Exception {
- // This is needed since sometimes some buckets are never picked in getPartition() and end up with 0 records
- if (!recordItr.hasNext()) {
- LOG.info("Empty partition");
- return Collections.singletonList((List<WriteStatus>) Collections.EMPTY_LIST).iterator();
- }
return new FlinkLazyInsertIterable<>(recordItr, true, config, instantTime, table, idPfx,
taskContextSupplier, new ExplicitWriteHandleFactory<>(writeHandle));
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
index 5f4b43a81ac4..85b85007056b 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
@@ -488,6 +488,14 @@ public class StreamWriteFunction extends AbstractStreamWriteFunction<HoodieFlink
* Write function to trigger the actual write action.
*/
protected interface WriteFunction extends Serializable {
- List<WriteStatus> write(Iterator<HoodieRecord> records, BucketInfo bucketInfo, String instant);
+ List<WriteStatus> doWrite(Iterator<HoodieRecord> records, BucketInfo bucketInfo, String instant);
+
+ default List<WriteStatus> write(Iterator<HoodieRecord> records, BucketInfo bucketInfo, String instant) {
+ if (!records.hasNext()) {
+ LOG.info("Empty records with bucket info => {}.", bucketInfo);
+ return Collections.emptyList();
+ }
+ return doWrite(records, bucketInfo, instant);
+ }
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java
index e2aec39549f9..c6e18fa74624 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java
@@ -45,11 +45,11 @@ import org.apache.flink.types.Row;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
/**
* Factory for CollectTableSink.
@@ -61,7 +61,7 @@ public class CollectSinkTableFactory implements DynamicTableSinkFactory {
public static final String FACTORY_ID = "collect";
// global results to collect and query
- public static final Map<Integer, List<Row>> RESULT = new HashMap<>();
+ public static final Map<Integer, List<Row>> RESULT = new ConcurrentHashMap<>();
// options
private static final ConfigOption<Integer> SINK_EXPECTED_ROW_NUM =