public DataPartitionResp getOrCreateDataPartition(final GetOrCreateDataPartitionPlan req) {
    // Check if the related Databases exist
    for (final String database : req.getPartitionSlotsMap().keySet()) {
      if (!isDatabaseExist(database)) {
        return new DataPartitionResp(
            new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())
                .setMessage(
                    String.format(
                        "Create DataPartition failed because the database: %s is not exists",
                        database)),
            false,
            null);
      }
    }

    // After all the DataPartitions are allocated,
    // all the read requests about DataPartitionTable are parallel.
    DataPartitionResp resp = getDataPartition(req);
    if (resp.isAllPartitionsExist()) {
      return resp;
    }

    // We serialize the creation process of DataPartitions to
    // ensure that each DataPartition is created by a unique CreateDataPartitionReq.
    // Because the number of DataPartitions per database is limited
    // by the number of SeriesPartitionSlots,
    // the number of serialized CreateDataPartitionReqs is acceptable.
    synchronized (this) {
      // Here we should check again if the DataPartition
      // has been created by other threads to improve concurrent performance
      resp = getDataPartition(req);
      if (resp.isAllPartitionsExist()) {
        return resp;
      }

      // Here we check if the related Databases exist again,
      // due to we don't have a transaction mechanism.
      for (final String database : req.getPartitionSlotsMap().keySet()) {
        if (!isDatabaseExist(database)) {
          return new DataPartitionResp(
              new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())
                  .setMessage(
                      String.format(
                          "Create DataPartition failed because the database: %s does not exist",
                          database)),
              false,
              null);
        }
      }

      // Filter unassigned DataPartitionSlots
      Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap =
          partitionInfo.filterUnassignedDataPartitionSlots(req.getPartitionSlotsMap());

      // Here we ensure that each StorageGroup has at least one DataRegion.
      // And if some StorageGroups own too many slots, extend DataRegion for them.

      // Map<StorageGroup, unassigned SeriesPartitionSlot count>
      Map<String, Integer> unassignedDataPartitionSlotsCountMap = new ConcurrentHashMap<>();
      unassignedDataPartitionSlotsMap.forEach(
          (storageGroup, unassignedDataPartitionSlots) ->
              unassignedDataPartitionSlotsCountMap.put(
                  storageGroup, unassignedDataPartitionSlots.size()));
      TSStatus status =
          extendRegionGroupIfNecessary(
              unassignedDataPartitionSlotsCountMap, TConsensusGroupType.DataRegion);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // Return an error code if Region extension failed
        resp.setStatus(status);
        return resp;
      }

      Map<String, DataPartitionTable> assignedDataPartition;
      try {
        assignedDataPartition =
            getLoadManager().allocateDataPartition(unassignedDataPartitionSlotsMap);
      } catch (DatabaseNotExistsException | NoAvailableRegionGroupException e) {
        status = getConsensusManager().confirmLeader();
        if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
          // The allocation might fail due to leadership change
          resp.setStatus(status);
          return resp;
        }

        LOGGER.error("Create DataPartition failed because: ", e);
        if (e instanceof DatabaseNotExistsException) {
          resp.setStatus(
              new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())
                  .setMessage(e.getMessage()));
        } else {
          resp.setStatus(
              new TSStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())
                  .setMessage(e.getMessage()));
        }
        return resp;
      }

      // Cache allocating result only if the current ConfigNode still holds its leadership
      CreateDataPartitionPlan createPlan = new CreateDataPartitionPlan();
      createPlan.setAssignedDataPartition(assignedDataPartition);

      status = consensusWritePartitionResult(createPlan);
      if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        // The allocation might fail due to consensus error
        resp.setStatus(status);
        return resp;
      }
    }

    resp = getDataPartition(req);
    if (!resp.isAllPartitionsExist()) {
      // Count the fail rate
      AtomicInteger totalSlotNum = new AtomicInteger();
      req.getPartitionSlotsMap()
          .forEach(
              (database, partitionSlots) ->
                  partitionSlots.forEach(
                      (seriesPartitionSlot, timeSlotList) ->
                          totalSlotNum.addAndGet(timeSlotList.getTimePartitionSlots().size())));

      AtomicInteger unassignedSlotNum = new AtomicInteger();
      Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> unassignedDataPartitionSlotsMap =
          partitionInfo.filterUnassignedDataPartitionSlots(req.getPartitionSlotsMap());
      StringJoiner errDatabases = new StringJoiner(", ", "[", "]");
      unassignedDataPartitionSlotsMap.forEach(
          (database, unassignedDataPartitionSlots) -> {
            AtomicBoolean hasUnassignedSlot = new AtomicBoolean(false);
            unassignedDataPartitionSlots.forEach(
                (seriesPartitionSlot, timeSlotList) -> {
                  if (!timeSlotList.getTimePartitionSlots().isEmpty()) {
                    hasUnassignedSlot.set(true);
                    unassignedSlotNum.addAndGet(timeSlotList.getTimePartitionSlots().size());
                  }
                });
            if (hasUnassignedSlot.get()) {
              errDatabases.add(database);
            }
          });

      String errMsg =
          String.format(
              "Lacked %d/%d DataPartition allocation result when get or create data partitions for databases: %s",
              unassignedSlotNum.get(), totalSlotNum.get(), errDatabases);
      LOGGER.error(errMsg);
      resp.setStatus(
          new TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
      return resp;
    }
    return resp;
  }
Get DataPartition and create a new one if it does not exist.

@param req DataPartitionPlan with Map{@literal <}StorageGroupName, Map{@literal
    <}SeriesPartitionSlot, List{@literal <}TimePartitionSlot{@literal >}{@literal >}{@literal
    >}
@return DataPartitionResp with DataPartition and {@link TSStatus}. {@link
    TSStatusCode#SUCCESS_STATUS} if all process finish. {@link TSStatusCode#NO_ENOUGH_DATANODE}
    if the DataNodes is not enough to create new Regions. {@link
    TSStatusCode#DATABASE_NOT_EXIST} if some database does not exist.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java