private ISinkChannel createSinkChannel(
      TFragmentInstanceId localFragmentInstanceId,
      TEndPoint remoteEndpoint,
      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 sink handle to plan node {} of {} for {}",
          remotePlanNodeId,
          remoteFragmentInstanceId,
          localFragmentInstanceId);
    }

    return new SinkChannel(
        remoteEndpoint,
        remoteFragmentInstanceId,
        remotePlanNodeId,
        localPlanNodeId,
        localFragmentInstanceId,
        localMemoryManager,
        executorService,
        tsBlockSerdeFactory.get(),
        new ISinkChannelListenerImpl(
            localFragmentInstanceId, instanceContext, instanceContext::failed, cnt),
        mppDataExchangeServiceClientManager);
  }
As we know the upstream and downstream node of shared queue, we don't need to put it into the
sink map.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java