public synchronized ISourceHandle createLocalSourceHandleForFragment(
      TFragmentInstanceId localFragmentInstanceId,
      String localPlanNodeId,
      String remotePlanNodeId,
      TFragmentInstanceId remoteFragmentInstanceId,
      int index,
      IMPPDataExchangeManagerCallback<Throwable> onFailureCallback) {
    if (sourceHandles.containsKey(localFragmentInstanceId)
        && sourceHandles.get(localFragmentInstanceId).containsKey(localPlanNodeId)) {
      throw new IllegalStateException(
          "Source handle for plan node "
              + localPlanNodeId
              + " of "
              + localFragmentInstanceId
              + " exists.");
    }

    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "Create local source handle from {} for plan node {} of {}",
          remoteFragmentInstanceId,
          localPlanNodeId,
          localFragmentInstanceId);
    }

    SharedTsBlockQueue queue;
    ISinkHandle sinkHandle = shuffleSinkHandles.get(remoteFragmentInstanceId);
    if (sinkHandle != null) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Get SharedTsBlockQueue from local sink handle");
      }
      queue = ((LocalSinkChannel) (sinkHandle.getChannel(index))).getSharedTsBlockQueue();
    } else {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Create SharedTsBlockQueue");
      }
      queue =
          new SharedTsBlockQueue(
              remoteFragmentInstanceId, remotePlanNodeId, localMemoryManager, executorService);
    }
    LocalSourceHandle localSourceHandle =
        new LocalSourceHandle(
            localFragmentInstanceId,
            localPlanNodeId,
            queue,
            new SourceHandleListenerImpl(onFailureCallback));
    sourceHandles
        .computeIfAbsent(localFragmentInstanceId, key -> new ConcurrentHashMap<>())
        .put(localPlanNodeId, localSourceHandle);
    return localSourceHandle;
  }
As we know the upstream and downstream node of shared queue, we don't need to put it into the
sourceHandle map.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java