lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2415] Add more info log for flink streaming reader (#3642)

Browse Source
This commit is contained in:
Danny Chan
2021-09-12 10:00:17 +08:00
committed by GitHub
parent 6228b17a3d
commit 9d5c3e5cb9
1 changed files with 12 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java
View File

@@ -248,6 +248,13 @@ public class StreamReadMonitoringFunction
List<HoodieCommitMetadata> activeMetadataList = instants.stream() List<HoodieCommitMetadata> activeMetadataList = instants.stream()
.map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, commitTimeline)).collect(Collectors.toList()); .map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, commitTimeline)).collect(Collectors.toList());
List<HoodieCommitMetadata> archivedMetadataList = getArchivedMetadata(instantRange, commitTimeline, tableName); List<HoodieCommitMetadata> archivedMetadataList = getArchivedMetadata(instantRange, commitTimeline, tableName);
if (archivedMetadataList.size() > 0) {
LOG.warn(""
+ "--------------------------------------------------------------------------------\n"
+ "---------- caution: the reader has fall behind too much from the writer,\n"
+ "---------- tweak 'read.tasks' option to add parallelism of read tasks.\n"
+ "--------------------------------------------------------------------------------");
}
List<HoodieCommitMetadata> metadataList = archivedMetadataList.size() > 0 List<HoodieCommitMetadata> metadataList = archivedMetadataList.size() > 0
? mergeList(activeMetadataList, archivedMetadataList) ? mergeList(activeMetadataList, archivedMetadataList)
: activeMetadataList; : activeMetadataList;
@@ -288,6 +295,11 @@ public class StreamReadMonitoringFunction
} }
// update the issues instant time // update the issues instant time
this.issuedInstant = commitToIssue; this.issuedInstant = commitToIssue;
LOG.info(""
+ "------------------------------------------------------------\n"
+ "---------- consumed to instant: {}\n"
+ "------------------------------------------------------------",
commitToIssue);
} }
@Override @Override