(iotdb) branch master updated: Pipe: Revert "Pipe: stop pipe using restarting strategy to unpin the wal's reference count to avoid WAL stacking (#11971)" to avoid unnecessary pipe drop during subtask exception handling (#12031)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new b78a88002f1 Pipe: Revert "Pipe: stop pipe using restarting strategy to 
unpin the wal's reference count to avoid WAL stacking (#11971)" to avoid 
unnecessary pipe drop during subtask exception handling (#12031)
b78a88002f1 is described below

commit b78a88002f1c41044dd7be0b2471ff313038e179
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Wed Feb 7 15:18:48 2024 +0800

    Pipe: Revert "Pipe: stop pipe using restarting strategy to unpin the wal's 
reference count to avoid WAL stacking (#11971)" to avoid unnecessary pipe drop 
during subtask exception handling (#12031)
    
    This reverts commit d0928eb15f68e21d85218520864cd07e92e22ab6.
---
 .../commons/pipe/agent/task/PipeTaskAgent.java     | 31 +++++++++++++---------
 1 file changed, 18 insertions(+), 13 deletions(-)

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