private synchronized ISinkChannel createLocalSinkChannel(
      TFragmentInstanceId localFragmentInstanceId,
      TFragmentInstanceId remoteFragmentInstanceId,
      String remotePlanNodeId,
      String localPlanNodeId,
      // TODO: replace with callbacks to decouple MPPDataExchangeManager from
      // FragmentInstanceContext
      FragmentInstanceContext instanceContext,
      AtomicInteger cnt) {

    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "Create local sink handle to plan node {} of {} for {}",
          remotePlanNodeId,
          remoteFragmentInstanceId,
          localFragmentInstanceId);
    }

    SharedTsBlockQueue queue;
    Map<String, ISourceHandle> sourceHandleMap = sourceHandles.get(remoteFragmentInstanceId);
    LocalSourceHandle localSourceHandle =
        sourceHandleMap == null ? null : (LocalSourceHandle) sourceHandleMap.get(remotePlanNodeId);
    if (localSourceHandle != null) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Get SharedTsBlockQueue from local source handle");
      }
      queue = localSourceHandle.getSharedTsBlockQueue();
    } else {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Create SharedTsBlockQueue");
      }
      queue =
          new SharedTsBlockQueue(
              localFragmentInstanceId, localPlanNodeId, localMemoryManager, executorService);
    }

    return new LocalSinkChannel(
        localFragmentInstanceId,
        queue,
        new ISinkChannelListenerImpl(
            localFragmentInstanceId, instanceContext, instanceContext::failed, cnt));
  }
Each FI has only one ShuffleSinkHandle.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java