private void asyncRecoverDataRegion() throws StartupException {
    long startRecoverTime = System.currentTimeMillis();
    isReadyForNonReadWriteFunctions.set(false);
    isReadyForReadAndWrite.set(false);
    cachedThreadPool =
        IoTDBThreadPoolFactory.newCachedThreadPool(ThreadName.STORAGE_ENGINE_CACHED_POOL.getName());

    List<Future<Void>> futures = new LinkedList<>();
    asyncRecover(futures);

    // wait until wal is recovered
    if (!CONFIG.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
      try {
        WALRecoverManager.getInstance().recover();
      } catch (WALException e) {
        LOGGER.error("Fail to recover wal.", e);
      }
    }

    // operations after all data regions are recovered
    Thread recoverEndTrigger =
        new Thread(
            () -> {
              checkResults(futures, "StorageEngine failed to recover.");
              isReadyForReadAndWrite.set(true);
              LOGGER.info(
                  "Storage Engine recover cost: {}s.",
                  (System.currentTimeMillis() - startRecoverTime) / 1000);

              PipeDataNodeAgent.runtime()
                  .registerPeriodicalJob(
                      "StorageEngine#operateFlush",
                      () -> operateFlush(new TFlushReq()),
                      PipeConfig.getInstance().getPipeStorageEngineFlushTimeIntervalMs() / 1000);
            },
            ThreadName.STORAGE_ENGINE_RECOVER_TRIGGER.getName());
    recoverEndTrigger.start();
  }
block insertion if the insertion is rejected by memory control
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java