[PR] Pipe: When stopping the pipe, unpin the wal's reference count to avoid WAL stacking [iotdb]

["MiniSho (via GitHub)" <gi...@apache.org>]

MiniSho opened a new pull request, #11971:
URL: https://github.com/apache/iotdb/pull/11971

   ## Description
   Change the execution flow of the `stop pipe` : 
   1. drop the pipe task firstly 
   2. then create a new pipe task with the same pipeMeta.
   
   When the pipe is dropped and re-created with the same pipeMeta, it will re-trigger the collection of historical data, and most of the files it pins are tsfile files not wal, because the wal files are flushed. By taking advantage of this property, we can make sure that the subsequent pins are new files, not old wal files, thus avoiding the wal stacking.


-- 
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-unsubscribe@iotdb.apache.org

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

Re: [PR] Pipe: When stopping the pipe, unpin the wal's reference count to avoid WAL stacking [iotdb]

["SteveYurongSu (via GitHub)" <gi...@apache.org>]

SteveYurongSu commented on code in PR #11971:
URL: https://github.com/apache/iotdb/pull/11971#discussion_r1468811197


##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java:
##########
@@ -555,34 +555,29 @@ private void startPipe(String pipeName, long creationTime) {
   }
 
   protected void stopPipe(String pipeName, long creationTime) {
-    final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
-
-    if (!checkBeforeStopPipe(existedPipeMeta, pipeName, creationTime)) {
-      return;
-    }
+    final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
 
-    // Get pipe tasks
-    final Map<TConsensusGroupId, PipeTask> pipeTasks =
-        pipeTaskManager.getPipeTasks(existedPipeMeta.getStaticMeta());
-    if (pipeTasks == null) {
+    if (!checkBeforeStopPipe(pipeMeta, pipeName, creationTime)) {
       LOGGER.info(
-          "Pipe {} (creation time = {}) has already been dropped or has not been created. "
-              + "Skip stopping.",
-          pipeName,
-          creationTime);
+          "Stop Pipe: Pipe {} has already been dropped or has not been created. Skip stopping.",
+          pipeName);
       return;
     }
 
-    // Trigger stop() method for each pipe task by parallel stream
+    // 1. Drop the pipe task
     final long startTime = System.currentTimeMillis();
-    pipeTasks.values().parallelStream().forEach(PipeTask::stop);
+    dropPipe(pipeName);
+
+    // 2. Set pipe meta status to STOPPED
+    pipeMeta.getRuntimeMeta().getStatus().set(PipeStatus.STOPPED);
+
+    // 3. create a new pipe with the same pipeMeta
+    createPipe(pipeMeta);

Review Comment:
   ![image](https://github.com/apache/iotdb/assets/30497621/1bd0ae9a-12f2-41fb-b260-865b0d6442f0)



-- 
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-unsubscribe@iotdb.apache.org

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

Re: [PR] Pipe: stop pipe using restarting strategy to unpin the wal's reference count to avoid WAL stacking [iotdb]

["SteveYurongSu (via GitHub)" <gi...@apache.org>]

SteveYurongSu merged PR #11971:
URL: https://github.com/apache/iotdb/pull/11971


-- 
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-unsubscribe@iotdb.apache.org

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