private Future<FragInstanceDispatchResult> dispatchWrite(List<FragmentInstance> instances) {
    final List<TSStatus> dispatchFailures = new ArrayList<>();
    int replicaNum = 0;

    // 1. do not dispatch if the RegionReplicaSet is empty
    final List<FragmentInstance> shouldDispatch = new ArrayList<>();
    for (final FragmentInstance instance : instances) {
      if (instance.getHostDataNode() == null
          || Optional.ofNullable(instance.getRegionReplicaSet())
                  .map(TRegionReplicaSet::getDataNodeLocationsSize)
                  .orElse(0)
              == 0) {
        dispatchFailures.add(
            new TSStatus(TSStatusCode.PLAN_FAILED_NETWORK_PARTITION.getStatusCode()));
      } else {
        replicaNum =
            Math.max(replicaNum, instance.getRegionReplicaSet().getDataNodeLocationsSize());
        shouldDispatch.add(instance);
      }
    }

    try {
      // 2. try the dispatch
      final List<FailedFragmentInstanceWithStatus> failedInstances =
          dispatchWriteOnce(shouldDispatch);

      // 3. decide if we need retry (we may decide the retry condition instance-wise, if needed)
      final boolean shouldRetry =
          !failedInstances.isEmpty() && maxRetryDurationInNs > 0 && replicaNum > 1;
      if (!shouldRetry) {
        failedInstances.forEach(fi -> dispatchFailures.add(fi.getFailureStatus()));
      } else {
        // 4. retry the instance on other replicas
        final List<FragmentInstance> retryInstances =
            failedInstances.stream()
                .map(FailedFragmentInstanceWithStatus::getInstance)
                .collect(Collectors.toList());
        // here we only retry over each replica once
        final List<FailedFragmentInstanceWithStatus> failedAfterRetry =
            dispatchRetryWrite(retryInstances, replicaNum);
        failedAfterRetry.forEach(fi -> dispatchFailures.add(fi.getFailureStatus()));
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOGGER.error("Interrupted when dispatching write async", e);
      return immediateFuture(
          new FragInstanceDispatchResult(
              RpcUtils.getStatus(
                  TSStatusCode.INTERNAL_SERVER_ERROR, "Interrupted errors: " + e.getMessage())));
    }

    if (dispatchFailures.isEmpty()) {
      return immediateFuture(new FragInstanceDispatchResult(true));
    }
    if (instances.size() == 1) {
      return immediateFuture(new FragInstanceDispatchResult(dispatchFailures.get(0)));
    } else {
      List<TSStatus> failureStatusList = new ArrayList<>();
      for (TSStatus dataNodeFailure : dispatchFailures) {
        if (dataNodeFailure.getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
          failureStatusList.addAll(dataNodeFailure.getSubStatus());
        } else {
          failureStatusList.add(dataNodeFailure);
        }
      }
      return immediateFuture(new FragInstanceDispatchResult(RpcUtils.getStatus(failureStatusList)));
    }
  }
Entrypoint for dispatching write fragment instances.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java