protected TPipeTransferResp handleTransferHandshakeV2(final PipeTransferHandshakeV2Req req)
      throws IOException {
    // Reject to handshake if the receiver can not take clusterId from config node.
    final String clusterIdFromConfigNode = getClusterId();
    if (clusterIdFromConfigNode == null) {
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_HANDSHAKE_ERROR,
              "Receiver can not get clusterId from config node.");
      LOGGER.warn(
          "Receiver id = {}: Handshake failed, response status = {}.", receiverId.get(), status);
      return new TPipeTransferResp(status);
    }

    // Reject to handshake if the request does not contain sender's clusterId.
    final String clusterIdFromHandshakeRequest =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID);
    if (clusterIdFromHandshakeRequest == null) {
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_HANDSHAKE_ERROR, "Handshake request does not contain clusterId.");
      LOGGER.warn(
          "Receiver id = {}: Handshake failed, response status = {}.", receiverId.get(), status);
      return new TPipeTransferResp(status);
    }

    // Reject to handshake if the receiver and sender are from the same cluster.
    if (Objects.equals(clusterIdFromConfigNode, clusterIdFromHandshakeRequest)) {
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_HANDSHAKE_ERROR,
              String.format(
                  "Receiver and sender are from the same cluster %s.",
                  clusterIdFromHandshakeRequest));
      LOGGER.warn(
          "Receiver id = {}: Handshake failed, response status = {}.", receiverId.get(), status);
      return new TPipeTransferResp(status);
    }

    // Reject to handshake if the request does not contain timestampPrecision.
    final String timestampPrecision =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION);
    if (timestampPrecision == null) {
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_HANDSHAKE_ERROR,
              "Handshake request does not contain timestampPrecision.");
      LOGGER.warn(
          "Receiver id = {}: Handshake failed, response status = {}.", receiverId.get(), status);
      return new TPipeTransferResp(status);
    }

    final String usernameString =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME);
    if (usernameString != null) {
      username = usernameString;
    }
    final String passwordString =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD);
    if (passwordString != null) {
      password = passwordString;
    }
    final TSStatus status = loginIfNecessary();
    if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
      LOGGER.warn(
          "Receiver id = {}: Handshake failed because login failed, response status = {}.",
          receiverId.get(),
          status);
      return new TPipeTransferResp(status);
    } else {
      LOGGER.info("Receiver id = {}: User {} login successfully.", receiverId.get(), username);
    }

    final String shouldConvertDataTypeOnTypeMismatchString =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_CONVERT_ON_TYPE_MISMATCH);
    if (shouldConvertDataTypeOnTypeMismatchString != null) {
      shouldConvertDataTypeOnTypeMismatch =
          Boolean.parseBoolean(shouldConvertDataTypeOnTypeMismatchString);
    }

    final String loadTsFileStrategyString =
        req.getParams().get(PipeTransferHandshakeConstant.HANDSHAKE_KEY_LOAD_TSFILE_STRATEGY);
    if (loadTsFileStrategyString != null) {
      isUsingAsyncLoadTsFileStrategy.set(
          Objects.equals(
              PipeConnectorConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_ASYNC_VALUE,
              loadTsFileStrategyString));
    }

    validateTsFile.set(
        Boolean.parseBoolean(
            req.getParams()
                .getOrDefault(
                    PipeTransferHandshakeConstant.HANDSHAKE_KEY_VALIDATE_TSFILE, "true")));

    shouldMarkAsPipeRequest.set(
        Boolean.parseBoolean(
            req.getParams()
                .getOrDefault(
                    PipeTransferHandshakeConstant.HANDSHAKE_KEY_MARK_AS_PIPE_REQUEST, "true")));

    // Handle the handshake request as a v1 request.
    // Here we construct a fake "dataNode" request to valid from v1 validation logic, though
    // it may not require the actual type of the v1 request.
    return handleTransferHandshakeV1(
        new PipeTransferHandshakeV1Req() {
          @Override
          protected PipeRequestType getPlanType() {
            return PipeRequestType.HANDSHAKE_DATANODE_V1;
          }
        }.convertToTPipeTransferReq(timestampPrecision));
  }
{@link IoTDBFileReceiver} is the parent class of receiver on both configNode and DataNode,
handling all the logic of parallel file receiving.
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java