[To dev/1.3] Pipe: prevent historical extractor from forwarding TsFiles generated by pipe when isForwardingPipeRequests is false (#15845) (#15864)

Author: XNX02

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java

@@ -66,6 +66,8 @@
 import java.util.stream.Collectors;
 
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_END_TIME_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_ENABLE_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_END_TIME_KEY;
@@ -79,6 +81,7 @@
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_MODS_ENABLE_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_START_TIME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_END_TIME_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_HISTORY_ENABLE_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_HISTORY_END_TIME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY;
@@ -119,6 +122,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
 
   private boolean isTerminateSignalSent = false;
 
+  private boolean isForwardingPipeRequests;
+
   private volatile boolean hasBeenStarted = false;
 
   private Queue<TsFileResource> pendingQueue;
@@ -330,9 +335,15 @@ public void customize(
                 || // Should extract deletion
                 listeningOptionPair.getRight());
 
+    isForwardingPipeRequests =
+        parameters.getBooleanOrDefault(
+            Arrays.asList(
+                EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY, SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
+            EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
+
     if (LOGGER.isInfoEnabled()) {
       LOGGER.info(
-          "Pipe {}@{}: historical data extraction time range, start time {}({}), end time {}({}), sloppy pattern {}, sloppy time range {}, should transfer mod file {}",
+          "Pipe {}@{}: historical data extraction time range, start time {}({}), end time {}({}), sloppy pattern {}, sloppy time range {}, should transfer mod file {}, is forwarding pipe requests: {}",
           pipeName,
           dataRegionId,
           DateTimeUtils.convertLongToDate(historicalDataExtractionStartTime),
@@ -341,7 +352,8 @@ public void customize(
           historicalDataExtractionEndTime,
           sloppyPattern,
           sloppyTimeRange,
-          shouldTransferModFile);
+          shouldTransferModFile,
+          isForwardingPipeRequests);
     }
   }
 
@@ -444,6 +456,9 @@ public synchronized void start() {
                     resource ->
                         // Some resource is marked as deleted but not removed from the list.
                         !resource.isDeleted()
+                            // Some resource is generated by pipe. We ignore them if the pipe should
+                            // not transfer pipe requests.
+                            && (!resource.isGeneratedByPipe() || isForwardingPipeRequests)
                             && (
                             // Some resource may not be closed due to the control of
                             // PIPE_MIN_FLUSH_INTERVAL_IN_MS. We simply ignore them.
@@ -461,6 +476,9 @@ && mayTsFileResourceOverlappedWithPattern(resource)))
                     resource ->
                         // Some resource is marked as deleted but not removed from the list.
                         !resource.isDeleted()
+                            // Some resource is generated by pipe. We ignore them if the pipe should
+                            // not transfer pipe requests.
+                            && (!resource.isGeneratedByPipe() || isForwardingPipeRequests)
                             && (
                             // Some resource may not be closed due to the control of
                             // PIPE_MIN_FLUSH_INTERVAL_IN_MS. We simply ignore them.