This is an automated email from the ASF dual-hosted git repository.
vinoyang 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 63fa2b6 [HUDI-1836] Logging consuming instant to
StreamReadOperator#processSplits (#2867)
63fa2b6 is described below
commit 63fa2b618626b2b8e9530ec105e0d4995eebb643
Author: hiscat <[email protected]>
AuthorDate: Tue Apr 27 14:00:59 2021 +0800
[HUDI-1836] Logging consuming instant to StreamReadOperator#processSplits
(#2867)
---
.../java/org/apache/hudi/source/StreamReadOperator.java | 3 +++
.../hudi/table/format/mor/MergeOnReadInputSplit.java | 15 +++++++++++++++
2 files changed, 18 insertions(+)
diff --git
a/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java
b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java
index b974c7f..86372e3 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java
@@ -152,6 +152,9 @@ public class StreamReadOperator extends
AbstractStreamOperator<RowData>
return;
}
+ // This log is important to indicate the consuming process, there is only
one log message for one data bucket.
+ LOG.info("Processing input split : {}", split);
+
format.open(split);
try {
RowData nextElement = null;
diff --git
a/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java
b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java
index 9d479e2..b929e7b 100644
---
a/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java
+++
b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java
@@ -93,4 +93,19 @@ public class MergeOnReadInputSplit implements InputSplit {
public int getSplitNumber() {
return this.splitNum;
}
+
+ @Override
+ public String toString() {
+ return "MergeOnReadInputSplit{"
+ + "splitNum=" + splitNum
+ + ", basePath=" + basePath
+ + ", logPaths=" + logPaths
+ + ", latestCommit='" + latestCommit + '\''
+ + ", tablePath='" + tablePath + '\''
+ + ", maxCompactionMemoryInBytes=" + maxCompactionMemoryInBytes
+ + ", mergeType='" + mergeType + '\''
+ + ", instantRange=" + instantRange
+ + '}';
+ }
+
}