Re: [PR] Subscription: avoid the file reading when the memory is not enough for tsfile slicing & implement all-out effort global timeout control & support client resume from breakpoint for tsfile consumption [iotdb]

[via GitHub]

VGalaxies commented on code in PR #13992:
URL: https://github.com/apache/iotdb/pull/13992#discussion_r1833952486


##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java:
##########
@@ -496,30 +501,41 @@ public Iterable<TabletInsertionEvent> 
toTabletInsertionEvents() {
     }
   }
 
-  private void waitForResourceEnough4Parsing() throws InterruptedException {
+  private void waitForResourceEnough4Parsing(final long timeoutMs) throws 
InterruptedException {
     final PipeMemoryManager memoryManager = 
PipeDataNodeResourceManager.memory();
     if (memoryManager.isEnough4TabletParsing()) {
       return;
     }
 
+    final long startTime = System.currentTimeMillis();
+    long lastRecordTime = startTime;
+
     final long memoryCheckIntervalMs =
         
PipeConfig.getInstance().getPipeTsFileParserCheckMemoryEnoughIntervalMs();
-    final long startTime = System.currentTimeMillis();
     while (!memoryManager.isEnough4TabletParsing()) {
       Thread.sleep(memoryCheckIntervalMs);
-    }
 
-    final double waitTimeSeconds = (System.currentTimeMillis() - startTime) / 
1000.0;
-    if (waitTimeSeconds > 1.0) {
-      LOGGER.info(
-          "Wait for resource enough for parsing {} for {} seconds.",
-          resource != null ? resource.getTsFilePath() : "tsfile",
-          waitTimeSeconds);
-    } else if (LOGGER.isDebugEnabled()) {
-      LOGGER.debug(
-          "Wait for resource enough for parsing {} for {} seconds.",
-          resource != null ? resource.getTsFilePath() : "tsfile",
-          waitTimeSeconds);
+      final long currentTime = System.currentTimeMillis();
+      final double elapsedRecordTimeSeconds = (currentTime - lastRecordTime) / 
1000.0;
+      final double waitTimeSeconds = (currentTime - startTime) / 1000.0;
+      if (elapsedRecordTimeSeconds > 1.0) {
+        LOGGER.info(
+            "Wait for resource enough for parsing {} for {} seconds.",
+            resource != null ? resource.getTsFilePath() : "tsfile",
+            waitTimeSeconds);
+        lastRecordTime = currentTime;
+      } else if (LOGGER.isDebugEnabled()) {
+        LOGGER.debug(
+            "Wait for resource enough for parsing {} for {} seconds.",
+            resource != null ? resource.getTsFilePath() : "tsfile",
+            waitTimeSeconds);
+      }
+
+      if (waitTimeSeconds * 1000 > timeoutMs) {
+        // should contain 'TimeoutException' in exception message
+        throw new InterruptedException(
+            String.format("TimeoutException: Waited %s seconds", 
waitTimeSeconds));
+      }

Review Comment:
   10s print once
   finally print



-- 
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: reviews-unsubscr...@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org