@Override
    public void run() {
      try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug(
              "[SendCloseSinkChannelEvent] to [ShuffleSinkHandle: {}, index: {}]).",
              remoteFragmentInstanceId,
              indexOfUpstreamSinkHandle);
        }
        int attempt = 0;
        TCloseSinkChannelEvent closeSinkChannelEvent =
            new TCloseSinkChannelEvent(remoteFragmentInstanceId, indexOfUpstreamSinkHandle);
        while (attempt < MAX_ATTEMPT_TIMES) {
          attempt += 1;
          try (SyncDataNodeMPPDataExchangeServiceClient client =
              mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
            client.onCloseSinkChannelEvent(closeSinkChannelEvent);
            break;
          } catch (Throwable e) {
            LOGGER.warn(
                "[SendCloseSinkChannelEvent] to [ShuffleSinkHandle: {}, index: {}] failed.).",
                remoteFragmentInstanceId,
                indexOfUpstreamSinkHandle);
            if (attempt == MAX_ATTEMPT_TIMES) {
              synchronized (SourceHandle.this) {
                sourceHandleListener.onFailure(SourceHandle.this, e);
              }
            }
            try {
              Thread.sleep(retryIntervalInMs);
            } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
              synchronized (SourceHandle.this) {
                sourceHandleListener.onFailure(SourceHandle.this, e);
              }
            }
          }
        }
      }
    }
Get data blocks from an upstream fragment instance.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java