func (s *KingbusServer) applyEntryNormal(e *raftpb.Entry) {
	if e.Type == raftpb.EntryNormal && len(e.Data) == 0 {
		return
	}
	dataType := e.Data[0]
	switch dataType {
	case utils.MySQLBinlogEventType:
		s.applyMySQLBinlogEvent(e)
		s.applyBroadcast.Send()
	case utils.NodeAttributesType:
		s.applyNodeAttributes(e)
	case utils.SyncerArgsType:
		s.applySyncerArgs(e)
	case utils.MasterInfoType:
		s.applyMasterInfo(e)
	case utils.MasterGtidPurgedType:
		s.applyMasterGtidPurged(e)
	default:
		log.Log.Fatalf("apply data not support,entry:%v,entry.data:%s", *e, string(e.Data))
	}
}
applyEntryNormal apples an EntryNormal type raftpb request to the EtcdServer
package server
import (
	"encoding/json"
	"fmt"
	"path"
	"github.com/flike/kingbus/storage/storagepb"
	uuid "github.com/satori/go.uuid"
	"github.com/flike/kingbus/config"
	"github.com/flike/kingbus/utils"
	"time"
	"github.com/coreos/etcd/pkg/pbutil"
	"github.com/coreos/etcd/pkg/types"
	etcdraft "github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/flike/kingbus/log"
	"github.com/flike/kingbus/mysql"
	"github.com/flike/kingbus/raft"
	"github.com/flike/kingbus/raft/membership"
	"github.com/flike/kingbus/storage"
	gomysql "github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/replication"
)
const (
	warnApplyDuration = 100 * time.Millisecond
)
func (s *KingbusServer) applyAll(ap *raft.ApplyEntry) {
	if len(ap.Entries) == 0 {
		ap.Notifyc <- struct{}{}
		return
	}
	s.applyEntries(ap)
	ap.Notifyc <- struct{}{}
}
func (s *KingbusServer) applyEntries(ap *raft.ApplyEntry) {
	if len(ap.Entries) == 0 {
		return
	}
	firstIndex := ap.Entries[0].Index
	appliedIndex := s.raftNode.GetAppliedIndex()
	if firstIndex > appliedIndex+1 {
		log.Log.Panicf("first index of committed entry[%d] should <= appliedIndex[%d] + 1", firstIndex, appliedIndex)
	}
	var ents []raftpb.Entry
	if appliedIndex+1-firstIndex < uint64(len(ap.Entries)) {
		ents = ap.Entries[appliedIndex+1-firstIndex:]
	}
	if len(ents) == 0 {
		return
	}
	var shouldstop bool
	if shouldstop = s.apply(ents); shouldstop {
		log.Log.Infof("the member has been permanently removed from the cluster")
		go s.Stop()
	}
}
type confChangeResponse struct {
	membs []*membership.Member
	err   error
}
func (s *KingbusServer) apply(es []raftpb.Entry) bool {
	var shouldStop bool
	for i := range es {
		e := es[i]
		switch e.Type {
		case raftpb.EntryNormal:
			s.applyEntryNormal(&e)
			s.raftNode.SetAppliedIndex(e.Index)
			s.raftNode.SetTerm(e.Term)
		case raftpb.EntryConfChange:
			var cc raftpb.ConfChange
			pbutil.MustUnmarshal(&cc, e.Data)
			removedSelf, err := s.applyConfChange(cc)
			s.raftNode.SetAppliedIndex(e.Index)
			s.raftNode.SetTerm(e.Term)
			shouldStop = shouldStop || removedSelf
			s.wait.Trigger(cc.ID, &confChangeResponse{s.cluster.Members(), err})
		default:
			log.Log.Panicf("entry type should be either EntryNormal or EntryConfChange")
		}
	}
	return shouldStop
}