private TPipeConsensusTransferResp handleTransferFilePiece(
      final PipeConsensusTransferFilePieceReq req, final boolean isSingleFile) {
    tsFilePieceReadWriteLock.readLock().lock();
    try {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(
            "PipeConsensus-PipeName-{}: starting to receive tsFile pieces", consensusPipeName);
      }
      long startBorrowTsFileWriterNanos = System.nanoTime();
      PipeConsensusTsFileWriter tsFileWriter =
          pipeConsensusTsFileWriterPool.borrowCorrespondingWriter(req.getCommitId());
      long startPreCheckNanos = System.nanoTime();
      pipeConsensusReceiverMetrics.recordBorrowTsFileWriterTimer(
          startPreCheckNanos - startBorrowTsFileWriterNanos);

      try {
        updateWritingFileIfNeeded(tsFileWriter, req.getFileName(), isSingleFile);
        final File writingFile = tsFileWriter.getWritingFile();
        final RandomAccessFile writingFileWriter = tsFileWriter.getWritingFileWriter();

        if (isWritingFileOffsetNonCorrect(tsFileWriter, req.getStartWritingOffset())) {
          if (!writingFile.getName().endsWith(TsFileConstant.TSFILE_SUFFIX)) {
            // If the file is a tsFile, then the content will not be changed for a specific
            // filename. However, for other files (mod, snapshot, etc.) the content varies for the
            // same name in different times, then we must rewrite the file to apply the newest
            // version.
            writingFileWriter.setLength(0);
          }

          final TSStatus status =
              RpcUtils.getStatus(
                  TSStatusCode.PIPE_CONSENSUS_TRANSFER_FILE_OFFSET_RESET,
                  String.format(
                      "Request sender to reset file reader's offset from %s to %s.",
                      req.getStartWritingOffset(), writingFileWriter.length()));
          LOGGER.warn(
              "PipeConsensus-PipeName-{}: File offset reset requested by receiver, response status = {}.",
              consensusPipeName,
              status);
          return PipeConsensusTransferFilePieceResp.toTPipeConsensusTransferResp(
              status, writingFileWriter.length());
        }

        long endPreCheckNanos = System.nanoTime();
        pipeConsensusReceiverMetrics.recordTsFilePiecePreCheckTime(
            endPreCheckNanos - startPreCheckNanos);
        writingFileWriter.write(req.getFilePiece());
        pipeConsensusReceiverMetrics.recordTsFilePieceWriteTime(
            System.nanoTime() - endPreCheckNanos);
        return PipeConsensusTransferFilePieceResp.toTPipeConsensusTransferResp(
            RpcUtils.SUCCESS_STATUS, writingFileWriter.length());
      } catch (Exception e) {
        LOGGER.warn(
            "PipeConsensus-PipeName-{}: Failed to write file piece from req {}.",
            consensusPipeName,
            req,
            e);
        final TSStatus status =
            RpcUtils.getStatus(
                TSStatusCode.PIPE_CONSENSUS_TRANSFER_FILE_ERROR,
                String.format("Failed to write file piece, because %s", e.getMessage()));
        try {
          return PipeConsensusTransferFilePieceResp.toTPipeConsensusTransferResp(
              status, PipeTransferFilePieceResp.ERROR_END_OFFSET);
        } catch (IOException ex) {
          return PipeConsensusTransferFilePieceResp.toTPipeConsensusTransferResp(status);
        } finally {
          // Exception may occur when disk system go wrong. At this time, we may reset all resource
          // and receive this file from scratch when leader will try to resend this file from
          // scratch
          // as well.
          closeCurrentWritingFileWriter(tsFileWriter, false);
          deleteCurrentWritingFile(tsFileWriter);
          // must return tsfileWriter after deleting its file.
          try {
            tsFileWriter.returnSelf(consensusPipeName);
          } catch (IOException | DiskSpaceInsufficientException returnException) {
            LOGGER.warn(
                "PipeConsensus-PipeName-{}: Failed to return tsFileWriter {}.",
                consensusPipeName,
                tsFileWriter,
                returnException);
          }
        }
      }
    } finally {
      tsFilePieceReadWriteLock.readLock().unlock();
    }
  }
This method cannot be set to synchronize. Receive events can be concurrent since reqBuffer but
load event must be synchronized.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/pipeconsensus/PipeConsensusReceiver.java