(iotdb) branch master updated: IoTV2: Close receiver's executor to prevent request stuck when stopping dn (#15505)

[jiangtian]

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

jiangtian 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 b18808abc92 IoTV2: Close receiver's executor to prevent request stuck 
when stopping dn (#15505)
b18808abc92 is described below

commit b18808abc9273fd6f853c452770b7a8baf987972
Author: Peng Junzhi <78788603+peng...@users.noreply.github.com>
AuthorDate: Thu May 15 14:22:19 2025 +0800

    IoTV2: Close receiver's executor to prevent request stuck when stopping dn 
(#15505)
    
    * close receiver's executor
    
    * typo
---
 .../protocol/pipeconsensus/PipeConsensusReceiver.java       |  7 +++++++
 .../protocol/pipeconsensus/PipeConsensusReceiverAgent.java  | 13 +++++++++++++
 .../org/apache/iotdb/db/service/DataNodeShutdownHook.java   |  4 ++++
 3 files changed, 24 insertions(+)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
index 92a1eec7306..9e5a41afc41 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java
@@ -1083,6 +1083,10 @@ public class PipeConsensusReceiver {
     LOGGER.info("Receiver-{} exit successfully.", 
consensusPipeName.toString());
   }
 
+  public void closeExecutor() {
+    requestExecutor.tryClose();
+  }
+
   private class PipeConsensusTsFileWriterPool {
     private final Lock lock = new ReentrantLock();
     private final List<PipeConsensusTsFileWriter> 
pipeConsensusTsFileWriterPool = new ArrayList<>();
@@ -1672,6 +1676,9 @@ public class PipeConsensusReceiver {
       try {
         isClosed.set(true);
       } finally {
+        // let all threads that may still await become active again to acquire 
lock instead of
+        // meaningless sleeping in the condition while lock is already 
released.
+        condition.signalAll();
         lock.unlock();
       }
     }
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java
index 92af030e623..21bd35868ff 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiverAgent.java
@@ -244,4 +244,17 @@ public class PipeConsensusReceiverAgent implements 
ConsensusPipeReceiver {
     consensusPipe2ReciverMap.clear();
     LOGGER.info("All Receivers related to {} are released.", dataRegionId);
   }
+
+  public final void closeReceiverExecutor() {
+    this.replicaReceiverMap.forEach(
+        (consensusGroupId, receiverMap) -> {
+          receiverMap.forEach(
+              (consensusPipeName, receiverReference) -> {
+                if (receiverReference != null) {
+                  receiverReference.get().closeExecutor();
+                  LOGGER.info("Receivers-{}' executor is closed.", 
consensusPipeName);
+                }
+              });
+        });
+  }
 }
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
index 34dfe8d35b7..8e5886ddd77 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.consensus.exception.ConsensusException;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
 import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
 import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
@@ -89,6 +90,9 @@ public class DataNodeShutdownHook extends Thread {
       triggerSnapshotForAllDataRegion();
     }
 
+    // Shutdown all consensus pipe's receiver
+    PipeDataNodeAgent.receiver().pipeConsensus().closeReceiverExecutor();
+
     // Actually stop all services started by the DataNode.
     // If we don't call this, services like the RestService are not stopped 
and I can't re-start
     // it.