public TSStatus createDataRegion(TRegionReplicaSet regionReplicaSet, String storageGroup) {
    TSStatus tsStatus;
    DataRegionId dataRegionId = new DataRegionId(regionReplicaSet.getRegionId().getId());
    try {
      storageEngine.createDataRegion(dataRegionId, storageGroup);
      dataRegionLockMap.put(dataRegionId, new ReentrantReadWriteLock(false));
      List<Peer> peers = new ArrayList<>();
      for (TDataNodeLocation dataNodeLocation : regionReplicaSet.getDataNodeLocations()) {
        TEndPoint endpoint =
            new TEndPoint(
                dataNodeLocation.getDataRegionConsensusEndPoint().getIp(),
                dataNodeLocation.getDataRegionConsensusEndPoint().getPort());
        peers.add(new Peer(dataRegionId, dataNodeLocation.getDataNodeId(), endpoint));
      }
      DataRegionConsensusImpl.getInstance().createLocalPeer(dataRegionId, peers);
      tsStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
    } catch (DataRegionException e) {
      LOGGER.error("Create Data Region {} failed because {}", storageGroup, e.getMessage());
      tsStatus = new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
      tsStatus.setMessage(String.format("Create Data Region failed because of %s", e.getMessage()));
    } catch (ConsensusGroupAlreadyExistException e) {
      tsStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
      tsStatus.setMessage(String.format("DataRegion %d already exists.", dataRegionId.getId()));
    } catch (ConsensusException e) {
      tsStatus = new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
      tsStatus.setMessage(e.getMessage());
    }
    return tsStatus;
  }
This class takes the responsibility of managing regions, executing PlanNode of write type on
according regions and controlling the execution concurrency.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java