fn propose_normal<T, C>(
        &mut self,
        poll_ctx: &mut PollContext<EK, ER, T, C>,
        mut req: RaftCmdRequest,
    ) -> Result<Either<u64, u64>> {
        if self.pending_merge_state.is_some()
            && req.get_admin_request().get_cmd_type() != AdminCmdType::RollbackMerge
        {
            return Err(box_err!(
                "{} peer in merging mode, can't do proposal.",
                self.tag
            ));
        }

        poll_ctx.raft_metrics.propose.normal += 1;

        if self.get_store().applied_index_term() == self.term() {
            // Only when applied index's term is equal to current leader's term, the information
            // in epoch checker is up to date and can be used to check epoch.
            if let Some(index) = self
                .cmd_epoch_checker
                .propose_check_epoch(&req, self.term())
            {
                return Ok(Either::Right(index));
            }
        } else if req.has_admin_request() {
            // The admin request is rejected because it may need to update epoch checker which
            // introduces an uncertainty and may breaks the correctness of epoch checker.
            return Err(box_err!(
                "{} peer has not applied to current term, applied_term {}, current_term {}",
                self.tag,
                self.get_store().applied_index_term(),
                self.term()
            ));
        }

        // TODO: validate request for unexpected changes.
        let ctx = match self.pre_propose(poll_ctx, &mut req) {
            Ok(ctx) => ctx,
            Err(e) => {
                warn!(
                    "skip proposal";
                    "region_id" => self.region_id,
                    "peer_id" => self.peer.get_id(),
                    "err" => ?e,
                    "error_code" => %e.error_code(),
                );
                return Err(e);
            }
        };

        let data = req.write_to_bytes()?;

        // TODO: use local histogram metrics
        PEER_PROPOSE_LOG_SIZE_HISTOGRAM.observe(data.len() as f64);

        if data.len() as u64 > poll_ctx.cfg.raft_entry_max_size.0 {
            error!(
                "entry is too large";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "size" => data.len(),
            );
            return Err(Error::RaftEntryTooLarge(self.region_id, data.len() as u64));
        }

        let propose_index = self.next_proposal_index();
        self.raft_group.propose(ctx.to_vec(), data)?;
        if self.next_proposal_index() == propose_index {
            // The message is dropped silently, this usually due to leader absence
            // or transferring leader. Both cases can be considered as NotLeader error.
            return Err(Error::NotLeader(self.region_id, None));
        }

        if ctx.contains(ProposalContext::PREPARE_MERGE) {
            self.last_proposed_prepare_merge_idx = propose_index;
        }

        Ok(Either::Left(propose_index))
    }
/// Propose normal request to raft
///
/// Returns Ok(Either::Left(index)) means the proposal is proposed successfully and is located on `index` position.
/// Ok(Either::Right(index)) means the proposal is rejected by `CmdEpochChecker` and the `index` is the position of
/// the last conflict admin cmd.
Propose normal request to raft
Returns Ok(Either::Left(index)) means the proposal is proposed successfully and is located on `index` position.
Ok(Either::Right(index)) means the proposal is rejected by `CmdEpochChecker` and the `index` is the position of
the last conflict admin cmd.
Propose normal request to raft Returns Ok ( Either :: Left ( index )) means the proposal is proposed successfully and is located on ` index ` position . Ok ( Either :: Right ( index )) means the proposal is rejected by ` CmdEpochChecker ` and the ` index ` is the position of the last conflict admin cmd .
fn propose_normal<T, C>(
        &mut self,
        poll_ctx: &mut PollContext<EK, ER, T, C>,
        mut req: RaftCmdRequest,
    ) -> Result<Either<u64, u64>> {
        if self.pending_merge_state.is_some()
            && req.get_admin_request().get_cmd_type() != AdminCmdType::RollbackMerge
        {
            return Err(box_err!(
                "{} peer in merging mode, can't do proposal.",
                self.tag
            ));
        }
        poll_ctx.raft_metrics.propose.normal += 1;
        if self.get_store().applied_index_term() == self.term() {
            // Only when applied index's term is equal to current leader's term, the information
            // in epoch checker is up to date and can be used to check epoch.
            if let Some(index) = self
                .cmd_epoch_checker
                .propose_check_epoch(&req, self.term())
            {
                return Ok(Either::Right(index));
            }
        } else if req.has_admin_request() {
            // The admin request is rejected because it may need to update epoch checker which
            // introduces an uncertainty and may breaks the correctness of epoch checker.
            return Err(box_err!(
                "{} peer has not applied to current term, applied_term {}, current_term {}",
                self.tag,
                self.get_store().applied_index_term(),
                self.term()
            ));
        }
        // TODO: validate request for unexpected changes.
        let ctx = match self.pre_propose(poll_ctx, &mut req) {
            Ok(ctx) => ctx,
            Err(e) => {
                warn!(
                    "skip proposal";
                    "region_id" => self.region_id,
                    "peer_id" => self.peer.get_id(),
                    "err" => ?e,
                    "error_code" => %e.error_code(),
                );
                return Err(e);
            }
        };
        let data = req.write_to_bytes()?;
        // TODO: use local histogram metrics
        PEER_PROPOSE_LOG_SIZE_HISTOGRAM.observe(data.len() as f64);
        if data.len() as u64 > poll_ctx.cfg.raft_entry_max_size.0 {
            error!(
                "entry is too large";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "size" => data.len(),
            );
            return Err(Error::RaftEntryTooLarge(self.region_id, data.len() as u64));
        }
        let propose_index = self.next_proposal_index();
        self.raft_group.propose(ctx.to_vec(), data)?;
        if self.next_proposal_index() == propose_index {
            // The message is dropped silently, this usually due to leader absence
            // or transferring leader. Both cases can be considered as NotLeader error.
            return Err(Error::NotLeader(self.region_id, None));
        }
        if ctx.contains(ProposalContext::PREPARE_MERGE) {
            self.last_proposed_prepare_merge_idx = propose_index;
        }
        Ok(Either::Left(propose_index))
    }
fn propose_normal < T , C > ( & mut self , poll_ctx : & mut PollContext < EK , ER , T , C > , mut req : RaftCmdRequest , ) -> Result < Either < u64 , u64 > > { if self . pending_merge_state . is_some ( ) && req . get_admin_request ( ) . get_cmd_type ( ) != AdminCmdType :: RollbackMerge { return Err ( box_err ! ( " " self tag ) ) ; } poll_ctx . raft_metrics . propose . normal += 1 ; if self . get_store ( ) . applied_index_term ( ) == self . term ( ) { // Only when applied index's term is equal to current leader's term, the information // in epoch checker is up to date and can be used to check epoch. if let Some ( index ) = self . cmd_epoch_checker . propose_check_epoch ( & req , self . term ( ) ) { return Ok ( Either :: Right ( index ) ) ; } } else if req . has_admin_request ( ) { // The admin request is rejected because it may need to update epoch checker which // introduces an uncertainty and may breaks the correctness of epoch checker. return Err ( box_err ! ( " " self tag self get_store ( ) applied_index_term ( ) self term ( ) ) ) ; } // TODO: validate request for unexpected changes. let ctx = match self . pre_propose ( poll_ctx , & mut req ) { Ok ( ctx ) => ctx , Err ( e ) => { warn ! ( " " " " self region_id " " self peer get_id ( ) " " e " " e error_code ( ) ) ; return Err ( e ) ; } } ; let data = req . write_to_bytes ( ) ? ; // TODO: use local histogram metrics PEER_PROPOSE_LOG_SIZE_HISTOGRAM . observe ( data . len ( ) as f64 ) ; if data . len ( ) as u64 > poll_ctx . cfg . raft_entry_max_size . 0 { error ! ( " " " " self region_id " " self peer get_id ( ) " " data len ( ) ) ; return Err ( Error :: RaftEntryTooLarge ( self . region_id , data . len ( ) as u64 ) ) ; } let propose_index = self . next_proposal_index ( ) ; self . raft_group . propose ( ctx . to_vec ( ) , data ) ? ; if self . next_proposal_index ( ) == propose_index { // The message is dropped silently, this usually due to leader absence // or transferring leader. Both cases can be considered as NotLeader error. return Err ( Error :: NotLeader ( self . region_id , None ) ) ; } if ctx . contains ( ProposalContext :: PREPARE_MERGE ) { self . last_proposed_prepare_merge_idx = propose_index ; } Ok ( Either :: Left ( propose_index ) ) }
Propose normal request to raft
Returns Ok(Either::Left(index)) means the proposal is proposed successfully and is located on `index` position.
Propose normal request to raft Returns Ok ( Either :: Left ( index )) means the proposal is proposed successfully and is located on ` index ` position .
&mut self
poll_ctx
&mut PollContext<EK, ER, T, C>
req
RaftCmdRequest