private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent)
      throws PipeException {
    final ProgressIndex progressIndex;
    final TPipeConsensusTransferResp resp;
    final TCommitId tCommitId =
        new TCommitId(
            pipeDeleteDataNodeEvent.getReplicateIndexForIoTV2(),
            pipeDeleteDataNodeEvent.getCommitterKey().getRestartTimes(),
            pipeDeleteDataNodeEvent.getRebootTimes());
    final TConsensusGroupId tConsensusGroupId =
        new TConsensusGroupId(TConsensusGroupType.DataRegion, consensusGroupId);

    try (final SyncPipeConsensusServiceClient syncPipeConsensusServiceClient =
        syncRetryClientManager.borrowClient(getFollowerUrl())) {
      progressIndex = pipeDeleteDataNodeEvent.getProgressIndex();
      resp =
          syncPipeConsensusServiceClient.pipeConsensusTransfer(
              PipeConsensusDeleteNodeReq.toTPipeConsensusTransferReq(
                  pipeDeleteDataNodeEvent.getDeleteDataNode(),
                  tCommitId,
                  tConsensusGroupId,
                  progressIndex,
                  thisDataNodeId));
    } catch (final Exception e) {
      throw new PipeRuntimeConnectorRetryTimesConfigurableException(
          String.format(
              PIPE_CONSENSUS_SYNC_CONNECTION_FAILED_FORMAT,
              getFollowerUrl().getIp(),
              getFollowerUrl().getPort(),
              DELETION_SCENARIO,
              e.getMessage()),
          Integer.MAX_VALUE);
    }

    final TSStatus status = resp.getStatus();
    // Only handle the failed statuses to avoid string format performance overhead
    if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
        && resp.getStatus().getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
      receiverStatusHandler.handle(
          status,
          String.format(
              "PipeConsensus transfer DeletionEvent %s error, result status %s.",
              pipeDeleteDataNodeEvent.getDeletionResource(), status),
          pipeDeleteDataNodeEvent.getDeleteDataNode().toString());
    }

    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "Successfully transferred deletion event {}.",
          pipeDeleteDataNodeEvent.getDeletionResource());
    }
  }
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