etcdserver/server.go
func (s *EtcdServer) apply(
	es []raftpb.Entry,
	confState *raftpb.ConfState,
) (appliedt uint64, appliedi uint64, shouldStop bool) {
	for i := range es {
		e := es[i]
		switch e.Type {
		case raftpb.EntryNormal:
			s.applyEntryNormal(&e)
			s.setAppliedIndex(e.Index)
			s.setTerm(e.Term)

		case raftpb.EntryConfChange:
			// set the consistent index of current executing entry
			if e.Index > s.consistIndex.ConsistentIndex() {
				s.consistIndex.setConsistentIndex(e.Index)
			}
			var cc raftpb.ConfChange
			pbutil.MustUnmarshal(&cc, e.Data)
			removedSelf, err := s.applyConfChange(cc, confState)
			s.setAppliedIndex(e.Index)
			s.setTerm(e.Term)
			shouldStop = shouldStop || removedSelf
			s.w.Trigger(cc.ID, &confChangeResponse{s.cluster.Members(), err})

		default:
			if lg := s.getLogger(); lg != nil {
				lg.Panic(
					"unknown entry type; must be either EntryNormal or EntryConfChange",
					zap.String("type", e.Type.String()),
				)
			} else {
				plog.Panicf("entry type should be either EntryNormal or EntryConfChange")
			}
		}
		appliedi, appliedt = e.Index, e.Term
	}
	return appliedt, appliedi, shouldStop
}
func (s *EtcdServer) apply(
	es []raftpb.Entry,
	confState *raftpb.ConfState,
) (appliedt uint64, appliedi uint64, shouldStop bool) {
	for i := range es {
		e := es[i]
		switch e.Type {
		case raftpb.EntryNormal:
			s.applyEntryNormal(&e)
			s.setAppliedIndex(e.Index)
			s.setTerm(e.Term)

		case raftpb.EntryConfChange:
			// set the consistent index of current executing entry
			if e.Index > s.consistIndex.ConsistentIndex() {
				s.consistIndex.setConsistentIndex(e.Index)
			}
			var cc raftpb.ConfChange
			pbutil.MustUnmarshal(&cc, e.Data)
			removedSelf, err := s.applyConfChange(cc, confState)
			s.setAppliedIndex(e.Index)
			s.setTerm(e.Term)
			shouldStop = shouldStop || removedSelf
			s.w.Trigger(cc.ID, &confChangeResponse{s.cluster.Members(), err})

		default:
			if lg := s.getLogger(); lg != nil {
				lg.Panic(
					"unknown entry type; must be either EntryNormal or EntryConfChange",
					zap.String("type", e.Type.String()),
				)
			} else {
				plog.Panicf("entry type should be either EntryNormal or EntryConfChange")
			}
		}
		appliedi, appliedt = e.Index, e.Term
	}
	return appliedt, appliedi, shouldStop
}
func ( s * EtcdServer ) apply ( es [ ] raftpb . Entry , confState * raftpb . ConfState , ) ( appliedt uint64 , appliedi uint64 , shouldStop bool ) { for i := range es { e := es [ i ] switch e . Type { case raftpb . EntryNormal : s . applyEntryNormal ( & e ) s . setAppliedIndex ( e . Index ) s . setTerm ( e . Term ) case raftpb . EntryConfChange : // set the consistent index of current executing entry if e . Index > s . consistIndex . ConsistentIndex ( ) { s . consistIndex . setConsistentIndex ( e . Index ) } var cc raftpb . ConfChange pbutil . MustUnmarshal ( & cc , e . Data ) removedSelf , err := s . applyConfChange ( cc , confState ) s . setAppliedIndex ( e . Index ) s . setTerm ( e . Term ) shouldStop = shouldStop || removedSelf s . w . Trigger ( cc . ID , & confChangeResponse { s . cluster . Members ( ) , err } ) default : if lg := s . getLogger ( ) ; lg != nil { lg . Panic ( " " , zap . String ( " " , e . Type . String ( ) ) , ) } else { plog . Panicf ( " " ) } } appliedi , appliedt = e . Index , e . Term } return appliedt , appliedi , shouldStop }
// apply takes entries received from Raft (after it has been committed) and
// applies them to the current state of the EtcdServer.
// The given entries should not be empty.
apply takes entries received from Raft ( after it has been committed ) and applies them to the current state of the EtcdServer . The given entries should not be empty .