@Override
    @SuppressWarnings("squid:S3776")
    public void run() {
      try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug(
              "[StartPullTsBlocksFromRemote] {}-{} [{}, {}) ",
              remoteFragmentInstanceId,
              indexOfUpstreamSinkHandle,
              startSequenceId,
              endSequenceId);
        }
        TGetDataBlockRequest req =
            new TGetDataBlockRequest(
                remoteFragmentInstanceId,
                startSequenceId,
                endSequenceId,
                indexOfUpstreamSinkHandle);
        int attempt = 0;
        while (attempt < MAX_ATTEMPT_TIMES) {
          attempt += 1;

          long startTime = System.nanoTime();
          try (SyncDataNodeMPPDataExchangeServiceClient client =
              mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
            TGetDataBlockResponse resp = client.getDataBlock(req);
            int tsBlockNum = resp.getTsBlocks().size();
            if (tsBlockNum == 0) {
              if (!closed) {
                // failed to pull TsBlocks
                LOGGER.warn(
                    "{} failed to pull TsBlocks [{}] to [{}] from SinkHandle {}, channel index {},",
                    localFragmentInstanceId,
                    startSequenceId,
                    endSequenceId,
                    remoteFragmentInstanceId,
                    indexOfUpstreamSinkHandle);
              }
              return;
            }
            List<ByteBuffer> tsBlocks = new ArrayList<>(tsBlockNum);
            tsBlocks.addAll(resp.getTsBlocks());

            if (LOGGER.isDebugEnabled()) {
              LOGGER.debug("[EndPullTsBlocksFromRemote] Count:{}", tsBlockNum);
            }
            DATA_EXCHANGE_COUNT_METRIC_SET.recordDataBlockNum(
                GET_DATA_BLOCK_NUM_CALLER, tsBlockNum);
            executorService.submit(
                new SendAcknowledgeDataBlockEventTask(startSequenceId, endSequenceId));
            synchronized (SourceHandle.this) {
              if (aborted || closed) {
                return;
              }
              for (int i = startSequenceId; i < endSequenceId; i++) {
                sequenceIdToTsBlock.put(i, tsBlocks.get(i - startSequenceId));
              }
              if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("[PutTsBlocksIntoBuffer]");
              }
              if (!blocked.isDone()) {
                blocked.set(null);
              }
            }
            break;
          } catch (Throwable e) {

            LOGGER.warn(
                "failed to get data block [{}, {}), attempt times: {}",
                startSequenceId,
                endSequenceId,
                attempt,
                e);

            // reach retry max times
            if (attempt == MAX_ATTEMPT_TIMES) {
              fail(e);
              return;
            }

            // sleep some time before retrying
            try {
              Thread.sleep(retryIntervalInMs);
            } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
              // if interrupted during sleeping, fast fail and don't retry any more
              fail(e);
            }
          } finally {
            DATA_EXCHANGE_COST_METRIC_SET.recordDataExchangeCost(
                GET_DATA_BLOCK_TASK_CALLER, System.nanoTime() - startTime);
          }
        }
      }
    }
Get data blocks from an upstream fragment instance.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java