IoTV2: Fix consensus pipe operation time out (#14399)

* fix consensus pipe operation time out

* add remove

Author: Pengzna

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java

@@ -59,7 +59,7 @@ public class SyncDataNodeClientPool {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SyncDataNodeClientPool.class);
 
-  private static final int DEFAULT_RETRY_NUM = 6;
+  private static final int DEFAULT_RETRY_NUM = 10;
 
   private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clientManager;
 

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java

@@ -74,7 +74,9 @@
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
 
 import static org.apache.iotdb.consensus.iot.IoTConsensus.getConsensusGroupIdsFromDir;
@@ -92,7 +94,9 @@ public class PipeConsensus implements IConsensus {
       new ConcurrentHashMap<>();
   private final PipeConsensusRPCService rpcService;
   private final RegisterManager registerManager = new RegisterManager();
-  private final ReentrantLock stateMachineMapLock = new ReentrantLock();
+  private final Map<ConsensusGroupId, ReentrantLock> consensusGroupIdReentrantLockMap =
+      new ConcurrentHashMap<>();
+  private final ReentrantReadWriteLock stateMachineMapLock = new ReentrantReadWriteLock();
   private final PipeConsensusConfig config;
   private final ConsensusPipeManager consensusPipeManager;
   private final ConsensusPipeGuardian consensusPipeGuardian;
@@ -190,7 +194,7 @@ private void checkAllConsensusPipe() {
                     entry -> entry.getKey().getConsensusGroupId(),
                     Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
     try {
-      stateMachineMapLock.lock();
+      stateMachineMapLock.writeLock().lock();
       stateMachineMap.forEach(
           (key, value) ->
               value.checkConsensusPipe(existedPipes.getOrDefault(key, ImmutableMap.of())));
@@ -214,7 +218,7 @@ private void checkAllConsensusPipe() {
                 }
               });
     } finally {
-      stateMachineMapLock.unlock();
+      stateMachineMapLock.writeLock().unlock();
     }
   }
 
@@ -263,8 +267,11 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
       throw new IllegalPeerEndpointException(thisNode, peers);
     }
 
+    Lock lock =
+        consensusGroupIdReentrantLockMap.computeIfAbsent(groupId, key -> new ReentrantLock());
     try {
-      stateMachineMapLock.lock();
+      lock.lock();
+      stateMachineMapLock.readLock().lock();
       if (stateMachineMap.containsKey(groupId)) {
         throw new ConsensusGroupAlreadyExistException(groupId);
       }
@@ -293,29 +300,36 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
       LOGGER.warn("Cannot create local peer for group {} with peers {}", groupId, peers, e);
       throw new ConsensusException(e);
     } finally {
-      stateMachineMapLock.unlock();
+      stateMachineMapLock.readLock().unlock();
+      lock.unlock();
     }
   }
 
   @Override
   public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException {
     KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.BEFORE_DELETE);
+    Lock lock =
+        consensusGroupIdReentrantLockMap.computeIfAbsent(groupId, key -> new ReentrantLock());
     try {
-      stateMachineMapLock.lock();
+      lock.lock();
+      stateMachineMapLock.readLock().lock();
       if (!stateMachineMap.containsKey(groupId)) {
         throw new ConsensusGroupNotExistException(groupId);
       }
 
       final PipeConsensusServerImpl consensus = stateMachineMap.get(groupId);
       consensus.clear();
+      stateMachineMap.remove(groupId);
 
       FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
       KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
     } catch (IOException e) {
       LOGGER.warn("Cannot delete local peer for group {}", groupId, e);
       throw new ConsensusException(e);
     } finally {
-      stateMachineMapLock.unlock();
+      stateMachineMapLock.readLock().unlock();
+      lock.unlock();
+      consensusGroupIdReentrantLockMap.remove(groupId);
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java

@@ -179,11 +179,11 @@ public TSStatus createNewRegion(ConsensusGroupId regionId, String storageGroup)
       if (regionId instanceof DataRegionId) {
         DataRegionId dataRegionId = (DataRegionId) regionId;
         storageEngine.createDataRegion(dataRegionId, storageGroup);
-        dataRegionLockMap.put(dataRegionId, new ReentrantReadWriteLock(false));
+        dataRegionLockMap.putIfAbsent(dataRegionId, new ReentrantReadWriteLock(false));
       } else {
         SchemaRegionId schemaRegionId = (SchemaRegionId) regionId;
         schemaEngine.createSchemaRegion(new PartialPath(storageGroup), schemaRegionId);
-        schemaRegionLockMap.put(schemaRegionId, new ReentrantReadWriteLock(false));
+        schemaRegionLockMap.putIfAbsent(schemaRegionId, new ReentrantReadWriteLock(false));
       }
     } catch (Exception e) {
       LOGGER.error("create new region {} error", regionId, e);