[IoTConsensusV2 X Pipe Deletion]: Release deletion resource when delete region & Improve for release when buffer is null (apache#14227)

* release deletion resource when delete region

* improve

Author: Pengzna

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java

@@ -41,6 +41,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.PriorityBlockingQueue;
@@ -142,7 +143,8 @@ public void start() {
   public boolean isAllDeletionFlushed() {
     buffersLock.lock();
     try {
-      return deletionResources.isEmpty() && serializeBuffer.position() == 0;
+      int pos = Optional.ofNullable(serializeBuffer).map(ByteBuffer::position).orElse(0);
+      return deletionResources.isEmpty() && pos == 0;
     } finally {
       buffersLock.unlock();
     }

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

@@ -83,6 +83,7 @@
 import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
 import org.apache.iotdb.db.exception.StorageEngineException;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
 import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
 import org.apache.iotdb.db.protocol.client.cn.DnToCnInternalServiceAsyncRequestManager;
 import org.apache.iotdb.db.protocol.client.cn.DnToCnRequestType;
@@ -2252,6 +2253,9 @@ public TSStatus deleteRegion(TConsensusGroupId tconsensusGroupId) {
     if (consensusGroupId instanceof DataRegionId) {
       try {
         DataRegionConsensusImpl.getInstance().deleteLocalPeer(consensusGroupId);
+        Optional.ofNullable(
+                DeletionResourceManager.getInstance(String.valueOf(tconsensusGroupId.getId())))
+            .ifPresent(DeletionResourceManager::close);
       } catch (ConsensusException e) {
         if (!(e instanceof ConsensusGroupNotExistException)) {
           return RpcUtils.getStatus(TSStatusCode.DELETE_REGION_ERROR, e.getMessage());