[HUDI-2415] Add more info log for flink streaming reader

apache · Sep 11, 2021 · e6d28ea · e6d28ea
1 parent 0624041
commit e6d28ea
Showing 1 changed file with 12 additions and 0 deletions.
diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java
@@ -248,6 +248,13 @@ public void monitorDirAndForwardSplits(SourceContext<MergeOnReadInputSplit> cont
     List<HoodieCommitMetadata> activeMetadataList = instants.stream()
         .map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, commitTimeline)).collect(Collectors.toList());
     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
         ? mergeList(activeMetadataList, archivedMetadataList)
         : activeMetadataList;
@@ -288,6 +295,11 @@ public void monitorDirAndForwardSplits(SourceContext<MergeOnReadInputSplit> cont
     }
     // update the issues instant time
     this.issuedInstant = commitToIssue;
+    LOG.info(""
+        + "------------------------------------------------------------\n"
+        + "---------- consumed to instant: {}\n"
+        + "------------------------------------------------------------",
+        commitToIssue);
   }
 
   @Override