Fix flush command stack problem and clean old code in DataRegion (apache#14027)

* Try to fix flush stack problem and clean old code in DataRegion

* clean code

* Fix flush command stuck

* fix ArrayIndexOutOfBoundsException

* refactor flush manager

* fix dataregion UT

Author: HTHou

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java

@@ -45,15 +45,13 @@
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.consensus.statemachine.dataregion.DataExecutionVisitor;
 import org.apache.iotdb.db.exception.DataRegionException;
 import org.apache.iotdb.db.exception.LoadReadOnlyException;
 import org.apache.iotdb.db.exception.StorageEngineException;
 import org.apache.iotdb.db.exception.TsFileProcessorException;
 import org.apache.iotdb.db.exception.WriteProcessRejectException;
 import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
 import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
 import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler;
 import org.apache.iotdb.db.service.metrics.WritingMetrics;
@@ -68,7 +66,6 @@
 import org.apache.iotdb.db.storageengine.dataregion.flush.FlushListener;
 import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
 import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy.DirectFlushPolicy;
-import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
 import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
 import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALException;
 import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
@@ -93,6 +90,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
@@ -234,7 +232,7 @@ private void asyncRecover(List<Future<Void>> futures) {
       for (DataRegionId dataRegionId : entry.getValue()) {
         Callable<Void> recoverDataRegionTask =
             () -> {
-              DataRegion dataRegion = null;
+              DataRegion dataRegion;
               try {
                 dataRegion = buildNewDataRegion(sgName, dataRegionId);
               } catch (DataRegionException e) {
@@ -268,7 +266,7 @@ public Map<String, List<DataRegionId>> getLocalDataRegionInfo() {
       }
       String sgName = sgDir.getName();
       List<DataRegionId> dataRegionIdList = new ArrayList<>();
-      for (File dataRegionDir : sgDir.listFiles()) {
+      for (File dataRegionDir : Objects.requireNonNull(sgDir.listFiles())) {
         if (!dataRegionDir.isDirectory()) {
           continue;
         }
@@ -446,21 +444,21 @@ public ServiceType getID() {
    * build a new data region
    *
    * @param dataRegionId data region id e.g. 1
-   * @param logicalStorageGroupName database name e.g. root.sg1
+   * @param databaseName database name e.g. root.sg1
    */
-  public DataRegion buildNewDataRegion(String logicalStorageGroupName, DataRegionId dataRegionId)
+  public DataRegion buildNewDataRegion(String databaseName, DataRegionId dataRegionId)
       throws DataRegionException {
     DataRegion dataRegion;
     LOGGER.info(
         "construct a data region instance, the database is {}, Thread is {}",
-        logicalStorageGroupName,
+        databaseName,
         Thread.currentThread().getId());
     dataRegion =
         new DataRegion(
-            systemDir + File.separator + logicalStorageGroupName,
+            systemDir + File.separator + databaseName,
             String.valueOf(dataRegionId.getId()),
             fileFlushPolicy,
-            logicalStorageGroupName);
+            databaseName);
     WRITING_METRICS.createFlushingMemTableStatusMetrics(dataRegionId);
     WRITING_METRICS.createDataRegionMemoryCostMetrics(dataRegion);
     WRITING_METRICS.createActiveMemtableCounterMetrics(dataRegionId);
@@ -469,11 +467,6 @@ public DataRegion buildNewDataRegion(String logicalStorageGroupName, DataRegionI
     return dataRegion;
   }
 
-  /** Write data into DataRegion. For standalone mode only. */
-  public TSStatus write(DataRegionId groupId, PlanNode planNode) {
-    return planNode.accept(new DataExecutionVisitor(), dataRegionMap.get(groupId));
-  }
-
   /** This function is just for unit test. */
   @TestOnly
   public synchronized void reset() {
@@ -513,29 +506,31 @@ public void forceCloseAllProcessor() throws TsFileProcessorException {
     checkResults(tasks, "Failed to force close processor.");
   }
 
-  public void closeStorageGroupProcessor(String storageGroupPath, boolean isSeq) {
+  public void syncCloseProcessorsInDatabase(String databaseName) {
     List<Future<Void>> tasks = new ArrayList<>();
     for (DataRegion dataRegion : dataRegionMap.values()) {
-      if (dataRegion.getDatabaseName().equals(storageGroupPath)) {
-        if (isSeq) {
-          for (TsFileProcessor tsFileProcessor : dataRegion.getWorkSequenceTsFileProcessors()) {
-            tasks.add(
-                cachedThreadPool.submit(
-                    () -> {
-                      dataRegion.syncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
-                      return null;
-                    }));
-          }
-        } else {
-          for (TsFileProcessor tsFileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
-            tasks.add(
-                cachedThreadPool.submit(
-                    () -> {
-                      dataRegion.syncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
-                      return null;
-                    }));
-          }
-        }
+      if (dataRegion != null && dataRegion.getDatabaseName().equals(databaseName)) {
+        tasks.add(
+            cachedThreadPool.submit(
+                () -> {
+                  dataRegion.syncCloseAllWorkingTsFileProcessors();
+                  return null;
+                }));
+      }
+    }
+    checkResults(tasks, "Failed to sync close processor.");
+  }
+
+  public void syncCloseProcessorsInDatabase(String databaseName, boolean isSeq) {
+    List<Future<Void>> tasks = new ArrayList<>();
+    for (DataRegion dataRegion : dataRegionMap.values()) {
+      if (dataRegion.getDatabaseName().equals(databaseName)) {
+        tasks.add(
+            cachedThreadPool.submit(
+                () -> {
+                  dataRegion.syncCloseWorkingTsFileProcessors(isSeq);
+                  return null;
+                }));
       }
     }
     checkResults(tasks, "Failed to close database processor.");
@@ -638,13 +633,12 @@ public void operateFlush(TFlushReq req) {
       StorageEngine.getInstance().syncCloseAllProcessor();
       WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
     } else {
-      for (String storageGroup : req.storageGroups) {
+      for (String databaseName : req.storageGroups) {
         if (req.isSeq == null) {
-          StorageEngine.getInstance().closeStorageGroupProcessor(storageGroup, true);
-          StorageEngine.getInstance().closeStorageGroupProcessor(storageGroup, false);
+          StorageEngine.getInstance().syncCloseProcessorsInDatabase(databaseName);
         } else {
           StorageEngine.getInstance()
-              .closeStorageGroupProcessor(storageGroup, Boolean.parseBoolean(req.isSeq));
+              .syncCloseProcessorsInDatabase(databaseName, Boolean.parseBoolean(req.isSeq));
         }
       }
     }
@@ -698,15 +692,15 @@ public TSStatus setConfiguration(TSetConfigurationReq req) {
 
   /**
    * Add a listener to listen flush start/end events. Notice that this addition only applies to
-   * TsFileProcessors created afterwards.
+   * TsFileProcessors created afterward.
    */
   public void registerFlushListener(FlushListener listener) {
     customFlushListeners.add(listener);
   }
 
   /**
    * Add a listener to listen file close events. Notice that this addition only applies to
-   * TsFileProcessors created afterwards.
+   * TsFileProcessors created afterward.
    */
   public void registerCloseFileListener(CloseFileListener listener) {
     customCloseFileListeners.add(listener);
@@ -723,24 +717,24 @@ private void makeSureNoOldRegion(DataRegionId regionId) {
 
   // When registering a new region, the coordinator needs to register the corresponding region with
   // the local storage before adding the corresponding consensusGroup to the consensus layer
-  public DataRegion createDataRegion(DataRegionId regionId, String sg) throws DataRegionException {
+  public void createDataRegion(DataRegionId regionId, String databaseName)
+      throws DataRegionException {
     makeSureNoOldRegion(regionId);
     AtomicReference<DataRegionException> exceptionAtomicReference = new AtomicReference<>(null);
-    DataRegion dataRegion =
-        dataRegionMap.computeIfAbsent(
-            regionId,
-            x -> {
-              try {
-                return buildNewDataRegion(sg, x);
-              } catch (DataRegionException e) {
-                exceptionAtomicReference.set(e);
-              }
-              return null;
-            });
+    dataRegionMap.computeIfAbsent(
+        regionId,
+        region -> {
+          try {
+            return buildNewDataRegion(databaseName, region);
+          } catch (DataRegionException e) {
+            exceptionAtomicReference.set(e);
+          }
+          return null;
+        });
+
     if (exceptionAtomicReference.get() != null) {
       throw exceptionAtomicReference.get();
     }
-    return dataRegion;
   }
 
   public void deleteDataRegion(DataRegionId regionId) {
@@ -966,7 +960,7 @@ public TSStatus executeLoadCommand(
     return status;
   }
 
-  /** reboot timed flush sequence/unsequence memetable thread */
+  /** reboot timed flush sequence/unsequence memtable thread */
   public void rebootTimedService() throws ShutdownException {
     LOGGER.info("Start rebooting all timed service.");