Pipe: Reduce degrade log from PipeRealtimeDataRegionHybridExtractor (apache#14888)

Author: SteveYurongSu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRegionExtractor;
 import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpoch;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
@@ -102,6 +103,11 @@ public void gcSchemaInfo() {
     device2Measurements = null;
   }
 
+  public boolean mayExtractorUseTablets(final PipeRealtimeDataRegionExtractor extractor) {
+    final TsFileEpoch.State state = tsFileEpoch.getState(extractor);
+    return state.equals(TsFileEpoch.State.EMPTY) || state.equals(TsFileEpoch.State.USING_TABLET);
+  }
+
   public void markAsTableModelEvent() {
     if (event instanceof PipeInsertionEvent) {
       ((PipeInsertionEvent) event).markAsTableModelEvent();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java

@@ -78,7 +78,7 @@ public boolean isNeedListenToInsertNode() {
   }
 
   private void extractTabletInsertion(final PipeRealtimeEvent event) {
-    if (canNotUseTabletAnyMore()) {
+    if (canNotUseTabletAnyMore(event)) {
       event
           .getTsFileEpoch()
           .migrateState(
@@ -201,7 +201,7 @@ private void extractTsFileInsertion(final PipeRealtimeEvent event) {
     }
   }
 
-  private boolean canNotUseTabletAnyMore() {
+  private boolean canNotUseTabletAnyMore(final PipeRealtimeEvent event) {
     // In the following 7 cases, we should not extract any more tablet events. all the data
     // represented by the tablet events should be carried by the following tsfile event:
     //  0. If the pipe task is currently restarted.
@@ -212,19 +212,19 @@ private boolean canNotUseTabletAnyMore() {
     //  4. The number of realtime tsfile events to transfer has exceeded the limit.
     //  5. The number of linked tsfiles has reached the dangerous threshold.
     //  6. The shallow memory usage of the insert node has reached the dangerous threshold.
-    return isPipeTaskCurrentlyRestarted()
-        || mayWalSizeReachThrottleThreshold()
-        || mayMemTablePinnedCountReachDangerousThreshold()
-        || isHistoricalTsFileEventCountExceededLimit()
-        || isRealtimeTsFileEventCountExceededLimit()
-        || mayTsFileLinkedCountReachDangerousThreshold()
-        || mayInsertNodeMemoryReachDangerousThreshold();
+    return isPipeTaskCurrentlyRestarted(event)
+        || mayWalSizeReachThrottleThreshold(event)
+        || mayMemTablePinnedCountReachDangerousThreshold(event)
+        || isHistoricalTsFileEventCountExceededLimit(event)
+        || isRealtimeTsFileEventCountExceededLimit(event)
+        || mayTsFileLinkedCountReachDangerousThreshold(event)
+        || mayInsertNodeMemoryReachDangerousThreshold(event);
   }
 
-  private boolean isPipeTaskCurrentlyRestarted() {
+  private boolean isPipeTaskCurrentlyRestarted(final PipeRealtimeEvent event) {
     final boolean isPipeTaskCurrentlyRestarted =
         PipeDataNodeAgent.task().isPipeTaskCurrentlyRestarted(pipeName);
-    if (isPipeTaskCurrentlyRestarted) {
+    if (isPipeTaskCurrentlyRestarted && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore1: Pipe task is currently restarted",
           pipeName,
@@ -233,11 +233,11 @@ private boolean isPipeTaskCurrentlyRestarted() {
     return isPipeTaskCurrentlyRestarted;
   }
 
-  private boolean mayWalSizeReachThrottleThreshold() {
+  private boolean mayWalSizeReachThrottleThreshold(final PipeRealtimeEvent event) {
     final boolean mayWalSizeReachThrottleThreshold =
         3 * WALManager.getInstance().getTotalDiskUsage()
             > IoTDBDescriptor.getInstance().getConfig().getThrottleThreshold();
-    if (mayWalSizeReachThrottleThreshold) {
+    if (mayWalSizeReachThrottleThreshold && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore2: Wal size {} has reached throttle threshold {}",
           pipeName,
@@ -248,11 +248,11 @@ private boolean mayWalSizeReachThrottleThreshold() {
     return mayWalSizeReachThrottleThreshold;
   }
 
-  private boolean mayMemTablePinnedCountReachDangerousThreshold() {
+  private boolean mayMemTablePinnedCountReachDangerousThreshold(final PipeRealtimeEvent event) {
     final boolean mayMemTablePinnedCountReachDangerousThreshold =
         PipeDataNodeResourceManager.wal().getPinnedWalCount()
             >= PipeConfig.getInstance().getPipeMaxAllowedPinnedMemTableCount();
-    if (mayMemTablePinnedCountReachDangerousThreshold) {
+    if (mayMemTablePinnedCountReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore3: The number of pinned memtables {} has reached the dangerous threshold {}",
           pipeName,
@@ -263,14 +263,14 @@ private boolean mayMemTablePinnedCountReachDangerousThreshold() {
     return mayMemTablePinnedCountReachDangerousThreshold;
   }
 
-  private boolean isHistoricalTsFileEventCountExceededLimit() {
+  private boolean isHistoricalTsFileEventCountExceededLimit(final PipeRealtimeEvent event) {
     final IoTDBDataRegionExtractor extractor =
         PipeDataRegionExtractorMetrics.getInstance().getExtractorMap().get(getTaskID());
     final boolean isHistoricalTsFileEventCountExceededLimit =
         Objects.nonNull(extractor)
             && extractor.getHistoricalTsFileInsertionEventCount()
                 >= PipeConfig.getInstance().getPipeMaxAllowedHistoricalTsFilePerDataRegion();
-    if (isHistoricalTsFileEventCountExceededLimit) {
+    if (isHistoricalTsFileEventCountExceededLimit && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore4: The number of historical tsFile events {} has exceeded the limit {}",
           pipeName,
@@ -281,11 +281,11 @@ private boolean isHistoricalTsFileEventCountExceededLimit() {
     return isHistoricalTsFileEventCountExceededLimit;
   }
 
-  private boolean isRealtimeTsFileEventCountExceededLimit() {
+  private boolean isRealtimeTsFileEventCountExceededLimit(final PipeRealtimeEvent event) {
     final boolean isRealtimeTsFileEventCountExceededLimit =
         pendingQueue.getTsFileInsertionEventCount()
             >= PipeConfig.getInstance().getPipeMaxAllowedPendingTsFileEpochPerDataRegion();
-    if (isRealtimeTsFileEventCountExceededLimit) {
+    if (isRealtimeTsFileEventCountExceededLimit && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore5: The number of realtime tsFile events {} has exceeded the limit {}",
           pipeName,
@@ -296,11 +296,11 @@ private boolean isRealtimeTsFileEventCountExceededLimit() {
     return isRealtimeTsFileEventCountExceededLimit;
   }
 
-  private boolean mayTsFileLinkedCountReachDangerousThreshold() {
+  private boolean mayTsFileLinkedCountReachDangerousThreshold(final PipeRealtimeEvent event) {
     final boolean mayTsFileLinkedCountReachDangerousThreshold =
         PipeDataNodeResourceManager.tsfile().getLinkedTsfileCount()
             >= PipeConfig.getInstance().getPipeMaxAllowedLinkedTsFileCount();
-    if (mayTsFileLinkedCountReachDangerousThreshold) {
+    if (mayTsFileLinkedCountReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore6: The number of linked tsfiles {} has reached the dangerous threshold {}",
           pipeName,
@@ -311,15 +311,15 @@ private boolean mayTsFileLinkedCountReachDangerousThreshold() {
     return mayTsFileLinkedCountReachDangerousThreshold;
   }
 
-  private boolean mayInsertNodeMemoryReachDangerousThreshold() {
+  private boolean mayInsertNodeMemoryReachDangerousThreshold(final PipeRealtimeEvent event) {
     final long floatingMemoryUsageInByte =
         PipeDataNodeAgent.task().getFloatingMemoryUsageInByte(pipeName);
     final long pipeCount = PipeDataNodeAgent.task().getPipeCount();
     final long freeMemorySizeInBytes =
         PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes();
     final boolean mayInsertNodeMemoryReachDangerousThreshold =
         3 * floatingMemoryUsageInByte * pipeCount >= 2 * freeMemorySizeInBytes;
-    if (mayInsertNodeMemoryReachDangerousThreshold) {
+    if (mayInsertNodeMemoryReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
       LOGGER.info(
           "Pipe task {}@{} canNotUseTabletAnyMore7: The shallow memory usage of the insert node {} has reached the dangerous threshold {}",
           pipeName,
@@ -379,7 +379,7 @@ private Event supplyTabletInsertion(final PipeRealtimeEvent event) {
                 return state;
               }
 
-              return canNotUseTabletAnyMore()
+              return canNotUseTabletAnyMore(event)
                   ? TsFileEpoch.State.USING_TSFILE
                   : TsFileEpoch.State.USING_TABLET;
             });