private void doTransfer(final PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
      throws PipeException {
    final File tsFile = pipeTsFileInsertionEvent.getTsFile();
    final File modFile = pipeTsFileInsertionEvent.getModFile();
    final TPipeConsensusTransferResp resp;

    try (final SyncPipeConsensusServiceClient syncPipeConsensusServiceClient =
        syncRetryClientManager.borrowClient(getFollowerUrl())) {
      final TCommitId tCommitId =
          new TCommitId(
              pipeTsFileInsertionEvent.getReplicateIndexForIoTV2(),
              pipeTsFileInsertionEvent.getCommitterKey().getRestartTimes(),
              pipeTsFileInsertionEvent.getRebootTimes());
      final TConsensusGroupId tConsensusGroupId =
          new TConsensusGroupId(TConsensusGroupType.DataRegion, consensusGroupId);

      // 1. Transfer tsFile, and mod file if exists
      if (pipeTsFileInsertionEvent.isWithMod()) {
        transferFilePieces(
            modFile, syncPipeConsensusServiceClient, true, tCommitId, tConsensusGroupId);
        transferFilePieces(
            tsFile, syncPipeConsensusServiceClient, true, tCommitId, tConsensusGroupId);
        // 2. Transfer file seal signal with mod, which means the file is transferred completely
        resp =
            syncPipeConsensusServiceClient.pipeConsensusTransfer(
                PipeConsensusTsFileSealWithModReq.toTPipeConsensusTransferReq(
                    modFile.getName(),
                    modFile.length(),
                    tsFile.getName(),
                    tsFile.length(),
                    pipeTsFileInsertionEvent.getFlushPointCount(),
                    tCommitId,
                    tConsensusGroupId,
                    pipeTsFileInsertionEvent.getProgressIndex(),
                    thisDataNodeId));
      } else {
        transferFilePieces(
            tsFile, syncPipeConsensusServiceClient, false, tCommitId, tConsensusGroupId);
        // 2. Transfer file seal signal without mod, which means the file is transferred completely
        resp =
            syncPipeConsensusServiceClient.pipeConsensusTransfer(
                PipeConsensusTsFileSealReq.toTPipeConsensusTransferReq(
                    tsFile.getName(),
                    tsFile.length(),
                    pipeTsFileInsertionEvent.getFlushPointCount(),
                    tCommitId,
                    tConsensusGroupId,
                    pipeTsFileInsertionEvent.getProgressIndex(),
                    thisDataNodeId));
      }
    } catch (final Exception e) {
      throw new PipeRuntimeConnectorRetryTimesConfigurableException(
          String.format(
              PIPE_CONSENSUS_SYNC_CONNECTION_FAILED_FORMAT,
              getFollowerUrl().getIp(),
              getFollowerUrl().getPort(),
              TSFILE_SCENARIO,
              e.getMessage()),
          Integer.MAX_VALUE);
    }

    final TSStatus status = resp.getStatus();
    // Only handle the failed statuses to avoid string format performance overhead
    if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
        && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
      receiverStatusHandler.handle(
          resp.getStatus(),
          String.format("Seal file %s error, result status %s.", tsFile, resp.getStatus()),
          tsFile.getName());
    }

    LOGGER.info("Successfully transferred file {}.", tsFile);
  }
This connector is used for PipeConsensus to transfer queued event.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusSyncConnector.java