private void asyncRecover(List<Future<Void>> futures) {
    Map<String, List<DataRegionId>> localDataRegionInfo = getLocalDataRegionInfo();
    localDataRegionInfo.values().forEach(list -> recoverDataRegionNum += list.size());
    readyDataRegionNum = new AtomicInteger(0);
    // init wal recover manager
    WALRecoverManager.getInstance()
        .setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(recoverDataRegionNum));
    for (Map.Entry<String, List<DataRegionId>> entry : localDataRegionInfo.entrySet()) {
      String sgName = entry.getKey();
      for (DataRegionId dataRegionId : entry.getValue()) {
        Callable<Void> recoverDataRegionTask =
            () -> {
              DataRegion dataRegion;
              try {
                dataRegion = buildNewDataRegion(sgName, dataRegionId);
              } catch (DataRegionException e) {
                LOGGER.error(
                    "Failed to recover data region {}[{}]", sgName, dataRegionId.getId(), e);
                return null;
              }
              dataRegionMap.put(dataRegionId, dataRegion);
              LOGGER.info(
                  "Data regions have been recovered {}/{}",
                  readyDataRegionNum.incrementAndGet(),
                  recoverDataRegionNum);
              return null;
            };
        futures.add(cachedThreadPool.submit(recoverDataRegionTask));
      }
    }
  }
block insertion if the insertion is rejected by memory control
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java