private List<FailedFragmentInstanceWithStatus> dispatchWriteOnce(List<FragmentInstance> instances)
      throws InterruptedException {
    if (instances.isEmpty()) {
      return Collections.emptyList();
    }

    final List<FragmentInstance> localInstances = new ArrayList<>();
    final List<FragmentInstance> remoteInstances = new ArrayList<>();
    for (FragmentInstance instance : instances) {
      if (isDispatchedToLocal(instance.getHostDataNode().getInternalEndPoint())) {
        localInstances.add(instance);
      } else {
        remoteInstances.add(instance);
      }
    }

    final List<FailedFragmentInstanceWithStatus> failedFragmentInstanceWithStatuses =
        new ArrayList<>();

    // 1. async dispatch to remote
    final AsyncPlanNodeSender asyncPlanNodeSender =
        new AsyncPlanNodeSender(asyncInternalServiceClientManager, remoteInstances);
    asyncPlanNodeSender.sendAll();

    // 2. sync dispatch to local
    if (!localInstances.isEmpty()) {
      long localScheduleStartTime = System.nanoTime();
      for (FragmentInstance localInstance : localInstances) {
        try (SetThreadName ignored = new SetThreadName(localInstance.getId().getFullId())) {
          dispatchLocally(localInstance);
        } catch (FragmentInstanceDispatchException e) {
          failedFragmentInstanceWithStatuses.add(
              new FailedFragmentInstanceWithStatus(localInstance, e.getFailureStatus()));
        } catch (Throwable t) {
          LOGGER.warn(DISPATCH_FAILED, t);
          failedFragmentInstanceWithStatuses.add(
              new FailedFragmentInstanceWithStatus(
                  localInstance,
                  RpcUtils.getStatus(
                      TSStatusCode.INTERNAL_SERVER_ERROR, UNEXPECTED_ERRORS + t.getMessage())));
        }
      }
      PERFORMANCE_OVERVIEW_METRICS.recordScheduleLocalCost(
          System.nanoTime() - localScheduleStartTime);
    }

    // 3. wait for remote dispatch results
    asyncPlanNodeSender.waitUntilCompleted();

    // 4. collect remote dispatch results
    failedFragmentInstanceWithStatuses.addAll(asyncPlanNodeSender.getFailedInstancesWithStatuses());

    return failedFragmentInstanceWithStatuses;
  }
Dispatch the given write instances once. It will dispatch the given instances locally or
remotely, give the host datanode.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java