Get old cluster info in raft storage

[greatsharp]

Reproduce steps:

1. Use raft as store, set only one peer in config-raft.yaml
2. Cluster version is 7 at present.
3. Add or delete a node in cluster to make the cluster version changed, new version is 8.
4. Log shows the cluster version changed correctly, but still get version 7 cluster info in syncClusterToNodes method.
5. Log shows version 7 cluster info was sent to nodes, not latest version 8.

[greatsharp]

[greatsharp]

Please apply PR #320 at first

[git-hulk]

@greatsharp Thanks for your fix.

[greatsharp]

@git-hulk how did you fixed this issue， I found it again in latest code。

Step 1: clusterStore.UpdateCluster invoke Set method in raft\node.go, it calls code n.raftNode.Propose(ctx, bytes) to commit new cluster topology.
Step 2: raft\node.go method runRaftMessages() calls applyDataEntry() in raft\store.go to update cluster topology in dataStore.

At this time, during step 1 and 2, if syncClusterToNodes method fetch cluster topology from dataStore, it will get old data.

[greatsharp]

Checked the etcd server method 'processInternalRaftRequestOnce'

the select logic after Propose is needed in kvrocks-controller

func (s *EtcdServer) processInternalRaftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (*apply2.Result, error) {
	ai := s.getAppliedIndex()
	ci := s.getCommittedIndex()
	if ci > ai+maxGapBetweenApplyAndCommitIndex {
		return nil, errors.ErrTooManyRequests
	}

	r.Header = &pb.RequestHeader{
		ID: s.reqIDGen.Next(),
	}

	// check authinfo if it is not InternalAuthenticateRequest
	if r.Authenticate == nil {
		authInfo, err := s.AuthInfoFromCtx(ctx)
		if err != nil {
			return nil, err
		}
		if authInfo != nil {
			r.Header.Username = authInfo.Username
			r.Header.AuthRevision = authInfo.Revision
		}
	}

	data, err := r.Marshal()
	if err != nil {
		return nil, err
	}

	if len(data) > int(s.Cfg.MaxRequestBytes) {
		return nil, errors.ErrRequestTooLarge
	}

	id := r.ID
	if id == 0 {
		id = r.Header.ID
	}
	ch := s.w.Register(id)

	cctx, cancel := context.WithTimeout(ctx, s.Cfg.ReqTimeout())
	defer cancel()

	start := time.Now()
	err = s.r.Propose(cctx, data)
	if err != nil {
		proposalsFailed.Inc()
		s.w.Trigger(id, nil) // GC wait
		return nil, err
	}
	proposalsPending.Inc()
	defer proposalsPending.Dec()

	select {
	case x := <-ch:
		return x.(*apply2.Result), nil
	case <-cctx.Done():
		proposalsFailed.Inc()
		s.w.Trigger(id, nil) // GC wait
		return nil, s.parseProposeCtxErr(cctx.Err(), start)
	case <-s.done:
		return nil, errors.ErrStopped
	}
}