Repository: matrixorigin/matrixone
Pull request #3321 (2022-06-24 05:22:18)

logservice: added ScheduleCommand handler
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Added schedule command handler required for achieving high availability.

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/logservice/commands.go b/pkg/logservice/commands.go
new file mode 100644
index 000000000..7cd49c7d1
--- /dev/null
+++ b/pkg/logservice/commands.go
@@ -0,0 +1,72 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
+)
+
+func (s *Service) handleCommands(cmds []hapb.ScheduleCommand) {
+	for _, cmd := range cmds {
+		switch cmd.ConfigChange.ChangeType {
+		case hapb.AddReplica:
+			s.handleAddReplica(cmd)
+		case hapb.RemoveReplica:
+			s.handleRemoveReplica(cmd)
+		case hapb.StartReplica:
+			s.handleStartReplica(cmd)
+		case hapb.StopReplica:
+			s.handleStopReplica(cmd)
+		default:
+			panic("unknown type")
+		}
+	}
+}
+
+func (s *Service) handleAddReplica(cmd hapb.ScheduleCommand) {
+	panic("not implemented due to the lack of target UUID in ScheduleCommand")
+}
+
+func (s *Service) handleRemoveReplica(cmd hapb.ScheduleCommand) {
+	shardID := cmd.ConfigChange.Replica.ShardID
+	replicaID := cmd.ConfigChange.Replica.ReplicaID
+	epoch := cmd.ConfigChange.Replica.Epoch
+	if err := s.store.removeReplica(shardID, replicaID, epoch); err != nil {
+		plog.Errorf("failed to remove replica %v", err)
+	}
+}
+
+func (s *Service) handleStartReplica(cmd hapb.ScheduleCommand) {
+	shardID := cmd.ConfigChange.Replica.ShardID
+	replicaID := cmd.ConfigChange.Replica.ReplicaID
+	if shardID == hakeeper.DefaultHAKeeperShardID {
+		if err := s.store.StartHAKeeperReplica(replicaID, nil, false); err != nil {
+			plog.Errorf("failed to start log replica %v", err)
+		}
+	} else {
+		if err := s.store.StartReplica(shardID, replicaID, nil, false); err != nil {
+			plog.Errorf("failed to start log replica %v", err)
+		}
+	}
+}
+
+func (s *Service) handleStopReplica(cmd hapb.ScheduleCommand) {
+	shardID := cmd.ConfigChange.Replica.ShardID
+	replicaID := cmd.ConfigChange.Replica.ReplicaID
+	if err := s.store.stopReplica(shardID, replicaID); err != nil {
+		plog.Errorf("failed to stop replica %v", err)
+	}
+}
diff --git a/pkg/logservice/hachecker.go b/pkg/logservice/hachecker.go
index d992c8ccb..7834c27e3 100644
--- a/pkg/logservice/hachecker.go
+++ b/pkg/logservice/hachecker.go
@@ -64,7 +64,7 @@ func (a *idAllocator) Capacity() uint64 {
 	return 0
 }
 
-func (l *logStore) updateIDAlloc(count uint64) error {
+func (l *store) updateIDAlloc(count uint64) error {
 	cmd := hakeeper.GetGetIDCmd(count)
 	ctx, cancel := context.WithTimeout(context.Background(), hakeeperDefaultTimeout)
 	defer cancel()
@@ -79,7 +79,7 @@ func (l *logStore) updateIDAlloc(count uint64) error {
 	return nil
 }
 
-func (l *logStore) healthCheck() {
+func (l *store) healthCheck() {
 	leaderID, term, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID)
 	if err != nil {
 		plog.Errorf("failed to get HAKeeper Leader ID, %v", err)
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index deedc81ec..3a2afd62b 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -43,7 +43,7 @@ type LogRecord = pb.LogRecord
 // be considered as the interface layer of the LogService.
 type Service struct {
 	cfg         Config
-	store       *logStore
+	store       *store
 	stopper     *syncutil.Stopper
 	connStopper *syncutil.Stopper
 }
diff --git a/pkg/logservice/shardinfo.go b/pkg/logservice/shardinfo.go
index 1dd0316e5..9ff374047 100644
--- a/pkg/logservice/shardinfo.go
+++ b/pkg/logservice/shardinfo.go
@@ -52,7 +52,7 @@ func (s *Service) GetShardInfo(shardID uint64) (ShardInfo, bool) {
 		if !ok {
 			return ShardInfo{}, false
 		}
-		var md logStoreMeta
+		var md storeMeta
 		md.unmarshal(data)
 		result.Replicas[nodeID] = ReplicaInfo{
 			UUID:           uuid,
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index ee83892fd..aac1fa806 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -40,20 +40,20 @@ var (
 	ErrOutOfRange = dragonboat.ErrInvalidRange
 )
 
-type logStoreMeta struct {
+type storeMeta struct {
 	serviceAddress string
 }
 
-func (l *logStoreMeta) marshal() []byte {
+func (l *storeMeta) marshal() []byte {
 	return []byte(l.serviceAddress)
 }
 
-func (l *logStoreMeta) unmarshal(data []byte) {
+func (l *storeMeta) unmarshal(data []byte) {
 	l.serviceAddress = string(data)
 }
 
 func getNodeHostConfig(cfg Config) config.NodeHostConfig {
-	meta := logStoreMeta{
+	meta := storeMeta{
 		serviceAddress: cfg.ServiceAddress,
 	}
 	return config.NodeHostConfig{
@@ -91,8 +91,8 @@ func getRaftConfig(shardID uint64, replicaID uint64) config.Config {
 	}
 }
 
-// logStore manages log shards including the HAKeeper shard on each node.
-type logStore struct {
+// store manages log shards including the HAKeeper shard on each node.
+type store struct {
 	cfg               Config
 	nh                *dragonboat.NodeHost
 	haKeeperReplicaID uint64
@@ -107,14 +107,14 @@ type logStore struct {
 	}
 }
 
-func newLogStore(cfg Config) (*logStore, error) {
+func newLogStore(cfg Config) (*store, error) {
 	plog.Infof("creating nodehost")
 	nh, err := dragonboat.NewNodeHost(getNodeHostConfig(cfg))
 	plog.Infof("create nodehost returned")
 	if err != nil {
 		return nil, err
 	}
-	ls := &logStore{
+	ls := &store{
 		cfg:     cfg,
 		nh:      nh,
 		alloc:   newIDAllocator(),
@@ -128,7 +128,7 @@ func newLogStore(cfg Config) (*logStore, error) {
 	return ls, nil
 }
 
-func (l *logStore) Close() error {
+func (l *store) Close() error {
 	l.stopper.Stop()
 	if l.nh != nil {
 		l.nh.Close()
@@ -136,17 +136,16 @@ func (l *logStore) Close() error {
 	return nil
 }
 
-func (l *logStore) ID() string {
+func (l *store) ID() string {
 	return l.nh.ID()
 }
 
-func (l *logStore) StartHAKeeperReplica(replicaID uint64,
-	initialReplicas map[uint64]dragonboat.Target) error {
+func (l *store) StartHAKeeperReplica(replicaID uint64,
+	initialReplicas map[uint64]dragonboat.Target, join bool) error {
 	l.haKeeperReplicaID = replicaID
 	raftConfig := getRaftConfig(hakeeper.DefaultHAKeeperShardID, replicaID)
-	// TODO: add another API for joining
 	if err := l.nh.StartReplica(initialReplicas,
-		false, hakeeper.NewStateMachine, raftConfig); err != nil {
+		join, hakeeper.NewStateMachine, raftConfig); err != nil {
 		return err
 	}
 	l.stopper.RunWorker(func() {
@@ -155,24 +154,70 @@ func (l *logStore) StartHAKeeperReplica(replicaID uint64,
 	return nil
 }
 
-func (l *logStore) StartReplica(shardID uint64, replicaID uint64,
-	initialReplicas map[uint64]dragonboat.Target) error {
+func (l *store) StartReplica(shardID uint64, replicaID uint64,
+	initialReplicas map[uint64]dragonboat.Target, join bool) error {
 	if shardID == hakeeper.DefaultHAKeeperShardID {
 		return ErrInvalidShardID
 	}
 	raftConfig := getRaftConfig(shardID, replicaID)
-	// TODO: add another API for joining
-	return l.nh.StartReplica(initialReplicas, false, newStateMachine, raftConfig)
+	return l.nh.StartReplica(initialReplicas, join, newStateMachine, raftConfig)
 }
 
-func (l *logStore) addReplica(shardID uint64, replicaID uint64,
+func (l *store) stopReplica(shardID uint64, replicaID uint64) error {
+	// FIXME: stop the hakeeper ticker when the replica being stopped
+	// is a hakeeper replica
+	return l.nh.StopReplica(shardID, replicaID)
+}
+
+func (l *store) addReplica(shardID uint64, replicaID uint64,
 	target dragonboat.Target, cci uint64) error {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	return l.nh.SyncRequestAddReplica(ctx, shardID, replicaID, target, cci)
+	count := 0
+	for {
+		count++
+		if err := l.nh.SyncRequestAddReplica(ctx, shardID, replicaID, target, cci); err != nil {
+			if errors.Is(err, dragonboat.ErrShardNotReady) {
+				l.retryWait()
+				continue
+			}
+			if errors.Is(err, dragonboat.ErrTimeoutTooSmall) && count > 1 {
+				return dragonboat.ErrTimeout
+			}
+			return err
+		}
+		return nil
+	}
+}
+
+func (l *store) removeReplica(shardID uint64, replicaID uint64, cci uint64) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	count := 0
+	for {
+		count++
+		if err := l.nh.SyncRequestDeleteReplica(ctx, shardID, replicaID, cci); err != nil {
+			if errors.Is(err, dragonboat.ErrShardNotReady) {
+				l.retryWait()
+				continue
+			}
+			if errors.Is(err, dragonboat.ErrTimeoutTooSmall) && count > 1 {
+				return dragonboat.ErrTimeout
+			}
+			return err
+		}
+		return nil
+	}
+}
+
+func (l *store) retryWait() {
+	if l.nh.NodeHostConfig().RTTMillisecond == 1 {
+		time.Sleep(time.Millisecond)
+	}
+	time.Sleep(time.Duration(l.nh.NodeHostConfig().RTTMillisecond/2) * time.Millisecond)
 }
 
-func (l *logStore) propose(ctx context.Context,
+func (l *store) propose(ctx context.Context,
 	session *cli.Session, cmd []byte) (sm.Result, error) {
 	count := 0
 	for {
@@ -180,7 +225,7 @@ func (l *logStore) propose(ctx context.Context,
 		result, err := l.nh.SyncPropose(ctx, session, cmd)
 		if err != nil {
 			if errors.Is(err, dragonboat.ErrShardNotReady) {
-				time.Sleep(time.Duration(l.nh.NodeHostConfig().RTTMillisecond) * time.Millisecond)
+				l.retryWait()
 				continue
 			}
 			if errors.Is(err, dragonboat.ErrTimeoutTooSmall) && count > 1 {
@@ -192,7 +237,7 @@ func (l *logStore) propose(ctx context.Context,
 	}
 }
 
-func (l *logStore) read(ctx context.Context,
+func (l *store) read(ctx context.Context,
 	shardID uint64, query interface{}) (interface{}, error) {
 	count := 0
 	for {
@@ -200,7 +245,7 @@ func (l *logStore) read(ctx context.Context,
 		result, err := l.nh.SyncRead(ctx, shardID, query)
 		if err != nil {
 			if errors.Is(err, dragonboat.ErrShardNotReady) {
-				time.Sleep(time.Duration(l.nh.NodeHostConfig().RTTMillisecond) * time.Millisecond)
+				l.retryWait()
 				continue
 			}
 			if errors.Is(err, dragonboat.ErrTimeoutTooSmall) && count > 1 {
@@ -212,7 +257,7 @@ func (l *logStore) read(ctx context.Context,
 	}
 }
 
-func (l *logStore) GetOrExtendDNLease(ctx context.Context,
+func (l *store) GetOrExtendDNLease(ctx context.Context,
 	shardID uint64, dnID uint64) error {
 	session := l.nh.GetNoOPSession(shardID)
 	cmd := getSetLeaseHolderCmd(dnID)
@@ -220,7 +265,7 @@ func (l *logStore) GetOrExtendDNLease(ctx context.Context,
 	return err
 }
 
-func (l *logStore) TruncateLog(ctx context.Context,
+func (l *store) TruncateLog(ctx context.Context,
 	shardID uint64, index Lsn) error {
 	session := l.nh.GetNoOPSession(shardID)
 	cmd := getSetTruncatedIndexCmd(index)
@@ -240,7 +285,7 @@ func (l *logStore) TruncateLog(ctx context.Context,
 	return nil
 }
 
-func (l *logStore) Append(ctx context.Context,
+func (l *store) Append(ctx context.Context,
 	shardID uint64, cmd []byte) (Lsn, error) {
 	if !isUserUpdate(cmd) {
 		panic(moerr.NewError(moerr.INVALID_INPUT, "not user update"))
@@ -262,7 +307,7 @@ func (l *logStore) Append(ctx context.Context,
 	return result.Value, nil
 }
 
-func (l *logStore) GetTruncatedIndex(ctx context.Context,
+func (l *store) GetTruncatedIndex(ctx context.Context,
 	shardID uint64) (uint64, error) {
 	v, err := l.read(ctx, shardID, truncatedIndexQuery{})
 	if err != nil {
@@ -271,7 +316,7 @@ func (l *logStore) GetTruncatedIndex(ctx context.Context,
 	return v.(uint64), nil
 }
 
-func (l *logStore) AddLogStoreHeartbeat(ctx context.Context,
+func (l *store) AddLogStoreHeartbeat(ctx context.Context,
 	hb pb.LogStoreHeartbeat) error {
 	data := MustMarshal(&hb)
 	cmd := hakeeper.GetLogStoreHeartbeatCmd(data)
@@ -283,7 +328,7 @@ func (l *logStore) AddLogStoreHeartbeat(ctx context.Context,
 	return nil
 }
 
-func (l *logStore) AddDNStoreHeartbeat(ctx context.Context,
+func (l *store) AddDNStoreHeartbeat(ctx context.Context,
 	hb pb.DNStoreHeartbeat) error {
 	data := MustMarshal(&hb)
 	cmd := hakeeper.GetDNStoreHeartbeatCmd(data)
@@ -295,7 +340,7 @@ func (l *logStore) AddDNStoreHeartbeat(ctx context.Context,
 	return nil
 }
 
-func (l *logStore) getLeaseHolderID(ctx context.Context,
+func (l *store) getLeaseHolderID(ctx context.Context,
 	shardID uint64, entries []raftpb.Entry) (uint64, error) {
 	if len(entries) == 0 {
 		panic("empty entries")
@@ -313,7 +358,7 @@ func (l *logStore) getLeaseHolderID(ctx context.Context,
 	return v.(uint64), nil
 }
 
-func (l *logStore) decodeCmd(e raftpb.Entry) []byte {
+func (l *store) decodeCmd(e raftpb.Entry) []byte {
 	if e.Type == raftpb.ApplicationEntry {
 		panic(moerr.NewError(moerr.INVALID_STATE, "unexpected entry type"))
 	}
@@ -333,7 +378,7 @@ func isRaftInternalEntry(e raftpb.Entry) bool {
 	return e.Type == raftpb.ConfigChangeEntry || e.Type == raftpb.MetadataEntry
 }
 
-func (l *logStore) filterEntries(ctx context.Context,
+func (l *store) filterEntries(ctx context.Context,
 	shardID uint64, entries []raftpb.Entry) ([]LogRecord, error) {
 	if len(entries) == 0 {
 		return []LogRecord{}, nil
@@ -375,7 +420,7 @@ func getNextIndex(entries []raftpb.Entry, firstIndex Lsn, lastIndex Lsn) Lsn {
 	return firstIndex
 }
 
-func (l *logStore) QueryLog(ctx context.Context, shardID uint64,
+func (l *store) QueryLog(ctx context.Context, shardID uint64,
 	firstIndex Lsn, maxSize uint64) ([]LogRecord, Lsn, error) {
 	v, err := l.read(ctx, shardID, indexQuery{})
 	if err != nil {
@@ -409,7 +454,7 @@ func (l *logStore) QueryLog(ctx context.Context, shardID uint64,
 	}
 }
 
-func (l *logStore) ticker() {
+func (l *store) ticker() {
 	ticker := time.NewTicker(hakeeper.TickDuration)
 	defer ticker.Stop()
 	haTicker := time.NewTicker(hakeeper.CheckDuration)
@@ -427,7 +472,7 @@ func (l *logStore) ticker() {
 	}
 }
 
-func (l *logStore) truncationWorker() {
+func (l *store) truncationWorker() {
 	for {
 		select {
 		case <-l.stopper.ShouldStop():
@@ -441,7 +486,7 @@ func (l *logStore) truncationWorker() {
 }
 
 // TODO: add test for this
-func (l *logStore) hakeeperTick() {
+func (l *store) hakeeperTick() {
 	leaderID, _, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID)
 	if err != nil {
 		plog.Errorf("failed to get HAKeeper Leader ID, %v", err)
@@ -460,7 +505,7 @@ func (l *logStore) hakeeperTick() {
 }
 
 // TODO: add tests for this
-func (l *logStore) truncateLog() error {
+func (l *store) truncateLog() error {
 	l.mu.Lock()
 	pendings := l.mu.pendingTruncate
 	l.mu.pendingTruncate = make(map[uint64]struct{})
@@ -502,7 +547,7 @@ func (l *logStore) truncateLog() error {
 	return nil
 }
 
-func (l *logStore) getHeartbeatMessage() pb.LogStoreHeartbeat {
+func (l *store) getHeartbeatMessage() pb.LogStoreHeartbeat {
 	m := pb.LogStoreHeartbeat{
 		UUID:           l.cfg.NodeHostID,
 		RaftAddress:    l.cfg.RaftAddress,


Test patch:
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 69f7ac9e1..8fc8e2465 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -42,7 +42,7 @@ func runClientTest(t *testing.T,
 
 	init := make(map[uint64]string)
 	init[2] = service.ID()
-	assert.NoError(t, service.store.StartReplica(1, 2, init))
+	assert.NoError(t, service.store.StartReplica(1, 2, init, false))
 
 	scfg := LogServiceClientConfig{
 		ReadOnly:         readOnly,
diff --git a/pkg/logservice/commands_test.go b/pkg/logservice/commands_test.go
new file mode 100644
index 000000000..1020b7d69
--- /dev/null
+++ b/pkg/logservice/commands_test.go
@@ -0,0 +1,94 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
+)
+
+func TestHandleStartReplica(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		cmd := hapb.ScheduleCommand{
+			ConfigChange: hapb.ConfigChange{
+				ChangeType: hapb.StartReplica,
+				Replica: hapb.Replica{
+					ShardID:   1,
+					ReplicaID: 1,
+				},
+			},
+		}
+		s.handleCommands([]hapb.ScheduleCommand{cmd})
+		mustHaveReplica(t, s.store, 1, 1)
+	}
+	runServiceTest(t, fn)
+}
+
+func TestHandleStopReplica(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		cmd := hapb.ScheduleCommand{
+			ConfigChange: hapb.ConfigChange{
+				ChangeType: hapb.StartReplica,
+				Replica: hapb.Replica{
+					ShardID:   1,
+					ReplicaID: 1,
+				},
+			},
+		}
+		s.handleCommands([]hapb.ScheduleCommand{cmd})
+		mustHaveReplica(t, s.store, 1, 1)
+
+		cmd = hapb.ScheduleCommand{
+			ConfigChange: hapb.ConfigChange{
+				ChangeType: hapb.StopReplica,
+				Replica: hapb.Replica{
+					ShardID:   1,
+					ReplicaID: 1,
+				},
+			},
+		}
+		s.handleCommands([]hapb.ScheduleCommand{cmd})
+		assert.False(t, hasReplica(s.store, 1, 1))
+	}
+	runServiceTest(t, fn)
+}
+
+func TestHandleRemoveReplica(t *testing.T) {
+	store1, store2, err := getTestStores()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, store1.Close())
+		require.NoError(t, store2.Close())
+	}()
+
+	service1 := Service{
+		store: store1,
+	}
+	cmd := hapb.ScheduleCommand{
+		ConfigChange: hapb.ConfigChange{
+			ChangeType: hapb.RemoveReplica,
+			Replica: hapb.Replica{
+				ShardID:   1,
+				ReplicaID: 2,
+			},
+		},
+	}
+	// TODO: add some checks
+	service1.handleCommands([]hapb.ScheduleCommand{cmd})
+}
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 7a27ec3a2..8e0a41547 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -50,7 +50,7 @@ func runServiceTest(t *testing.T, fn func(*testing.T, *Service)) {
 	require.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[1] = service.ID()
-	require.NoError(t, service.store.StartReplica(1, 1, peers))
+	require.NoError(t, service.store.StartReplica(1, 1, peers, false))
 	defer func() {
 		assert.NoError(t, service.Close())
 	}()
@@ -302,7 +302,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	}()
 	peers1 := make(map[uint64]dragonboat.Target)
 	peers1[1] = service1.ID()
-	assert.NoError(t, service1.store.StartReplica(1, 1, peers1))
+	assert.NoError(t, service1.store.StartReplica(1, 1, peers1, false))
 
 	service2, err := NewService(cfg2)
 	require.NoError(t, err)
@@ -311,7 +311,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	}()
 	peers2 := make(map[uint64]dragonboat.Target)
 	peers2[1] = service2.ID()
-	assert.NoError(t, service2.store.StartReplica(2, 1, peers2))
+	assert.NoError(t, service2.store.StartReplica(2, 1, peers2, false))
 
 	nhID1 := service1.ID()
 	nhID2 := service2.ID()
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index b0dcc0b6d..ff52c5bc9 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -21,11 +21,14 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/google/uuid"
 	"github.com/lni/dragonboat/v4"
 	"github.com/lni/goutils/leaktest"
 	"github.com/lni/vfs"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -72,14 +75,14 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	}()
 }
 
-func getTestStore(cfg Config) (*logStore, error) {
+func getTestStore(cfg Config) (*store, error) {
 	store, err := newLogStore(cfg)
 	if err != nil {
 		return nil, err
 	}
 	peers := make(map[uint64]dragonboat.Target)
 	peers[2] = store.nh.ID()
-	if err := store.StartReplica(1, 2, peers); err != nil {
+	if err := store.StartReplica(1, 2, peers, false); err != nil {
 		store.Close()
 		return nil, err
 	}
@@ -94,10 +97,11 @@ func TestHAKeeperCanBeStarted(t *testing.T) {
 	assert.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[2] = store.nh.ID()
-	assert.NoError(t, store.StartHAKeeperReplica(2, peers))
+	assert.NoError(t, store.StartHAKeeperReplica(2, peers, false))
 	defer func() {
 		assert.NoError(t, store.Close())
 	}()
+	mustHaveReplica(t, store, hakeeper.DefaultHAKeeperShardID, 2)
 }
 
 func TestStateMachineCanBeStarted(t *testing.T) {
@@ -109,9 +113,24 @@ func TestStateMachineCanBeStarted(t *testing.T) {
 	defer func() {
 		assert.NoError(t, store.Close())
 	}()
+	mustHaveReplica(t, store, 1, 2)
 }
 
-func runStoreTest(t *testing.T, fn func(*testing.T, *logStore)) {
+func TestReplicaCanBeStopped(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	cfg := getStoreTestConfig()
+	defer vfs.ReportLeakedFD(cfg.FS, t)
+	store, err := getTestStore(cfg)
+	assert.NoError(t, err)
+	defer func() {
+		assert.NoError(t, store.Close())
+	}()
+	mustHaveReplica(t, store, 1, 2)
+	store.stopReplica(1, 2)
+	assert.False(t, hasReplica(store, 1, 2))
+}
+
+func runStoreTest(t *testing.T, fn func(*testing.T, *store)) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
@@ -132,7 +151,7 @@ func getTestUserEntry() []byte {
 }
 
 func TestGetOrExtendLease(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		assert.NoError(t, store.GetOrExtendDNLease(ctx, 1, 100))
@@ -141,7 +160,7 @@ func TestGetOrExtendLease(t *testing.T) {
 }
 
 func TestAppendLog(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		assert.NoError(t, store.GetOrExtendDNLease(ctx, 1, 100))
@@ -154,7 +173,7 @@ func TestAppendLog(t *testing.T) {
 }
 
 func TestAppendLogIsRejectedForMismatchedLeaseHolderID(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		assert.NoError(t, store.GetOrExtendDNLease(ctx, 1, 100))
@@ -169,7 +188,7 @@ func TestAppendLogIsRejectedForMismatchedLeaseHolderID(t *testing.T) {
 }
 
 func TestTruncateLog(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		assert.NoError(t, store.GetOrExtendDNLease(ctx, 1, 100))
@@ -184,7 +203,7 @@ func TestTruncateLog(t *testing.T) {
 }
 
 func TestGetTruncatedIndex(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		index, err := store.GetTruncatedIndex(ctx, 1)
@@ -203,7 +222,7 @@ func TestGetTruncatedIndex(t *testing.T) {
 }
 
 func TestQueryLog(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		ctx, cancel := context.WithTimeout(context.Background(), testIOTimeout)
 		defer cancel()
 		assert.NoError(t, store.GetOrExtendDNLease(ctx, 1, 100))
@@ -240,21 +259,21 @@ func TestQueryLog(t *testing.T) {
 }
 
 func TestHAKeeperTick(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)
 		peers[1] = store.ID()
-		assert.NoError(t, store.StartHAKeeperReplica(1, peers))
+		assert.NoError(t, store.StartHAKeeperReplica(1, peers, false))
 		store.hakeeperTick()
 	}
 	runStoreTest(t, fn)
 }
 
 func TestGetHeartbeatMessage(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)
 		peers[1] = store.ID()
-		assert.NoError(t, store.StartReplica(10, 1, peers))
-		assert.NoError(t, store.StartHAKeeperReplica(1, peers))
+		assert.NoError(t, store.StartReplica(10, 1, peers, false))
+		assert.NoError(t, store.StartHAKeeperReplica(1, peers, false))
 
 		m := store.getHeartbeatMessage()
 		// hakeeper shard is included
@@ -264,10 +283,10 @@ func TestGetHeartbeatMessage(t *testing.T) {
 }
 
 func TestAddHeartbeat(t *testing.T) {
-	fn := func(t *testing.T, store *logStore) {
+	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)
 		peers[1] = store.ID()
-		assert.NoError(t, store.StartHAKeeperReplica(1, peers))
+		assert.NoError(t, store.StartHAKeeperReplica(1, peers, false))
 
 		m := store.getHeartbeatMessage()
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
@@ -283,3 +302,137 @@ func TestAddHeartbeat(t *testing.T) {
 	}
 	runStoreTest(t, fn)
 }
+
+func TestAddReplicaRejectedForInvalidCCI(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		err := store.addReplica(1, 100, uuid.New().String(), 0)
+		assert.Equal(t, dragonboat.ErrRejected, err)
+	}
+	runStoreTest(t, fn)
+}
+
+func TestAddReplica(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		for {
+			_, _, ok, err := store.nh.GetLeaderID(1)
+			require.NoError(t, err)
+			if ok {
+				break
+			}
+			time.Sleep(time.Millisecond)
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		m, err := store.nh.SyncGetShardMembership(ctx, 1)
+		require.NoError(t, err)
+		err = store.addReplica(1, 100, uuid.New().String(), m.ConfigChangeID)
+		assert.NoError(t, err)
+		hb := store.getHeartbeatMessage()
+		assert.Equal(t, 2, len(hb.Replicas[0].Replicas))
+	}
+	runStoreTest(t, fn)
+}
+
+func getTestStores() (*store, *store, error) {
+	cfg1 := Config{
+		FS:                  vfs.NewStrictMem(),
+		DeploymentID:        1,
+		RTTMillisecond:      5,
+		DataDir:             "data-1",
+		ServiceAddress:      "127.0.0.1:9001",
+		RaftAddress:         "127.0.0.1:9002",
+		GossipAddress:       "127.0.0.1:9011",
+		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
+	}
+	cfg1.Fill()
+	store1, err := newLogStore(cfg1)
+	if err != nil {
+		return nil, nil, err
+	}
+	cfg2 := Config{
+		FS:                  vfs.NewStrictMem(),
+		DeploymentID:        1,
+		RTTMillisecond:      5,
+		DataDir:             "data-1",
+		ServiceAddress:      "127.0.0.1:9006",
+		RaftAddress:         "127.0.0.1:9007",
+		GossipAddress:       "127.0.0.1:9012",
+		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
+	}
+	cfg2.Fill()
+	store2, err := newLogStore(cfg2)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	peers1 := make(map[uint64]dragonboat.Target)
+	peers1[1] = store1.nh.ID()
+	peers1[2] = store2.nh.ID()
+	if err := store1.StartReplica(1, 1, peers1, false); err != nil {
+		return nil, nil, err
+	}
+	peers2 := make(map[uint64]dragonboat.Target)
+	peers2[1] = store1.nh.ID()
+	peers2[2] = store2.nh.ID()
+	if err := store2.StartReplica(1, 2, peers2, false); err != nil {
+		return nil, nil, err
+	}
+
+	for {
+		_, _, ok, err := store1.nh.GetLeaderID(1)
+		if err != nil {
+			return nil, nil, err
+		}
+		if ok {
+			break
+		}
+		time.Sleep(time.Millisecond)
+	}
+	return store1, store2, nil
+}
+
+func TestRemoveReplica(t *testing.T) {
+	store1, store2, err := getTestStores()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, store1.Close())
+		require.NoError(t, store2.Close())
+	}()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	for {
+		m, err := store1.nh.SyncGetShardMembership(ctx, 1)
+		if err == dragonboat.ErrShardNotReady {
+			time.Sleep(time.Millisecond)
+			continue
+		}
+		require.NoError(t, err)
+		require.NoError(t, store1.removeReplica(1, 2, m.ConfigChangeID))
+		return
+	}
+}
+
+func hasReplica(s *store, shardID uint64, replicaID uint64) bool {
+	hb := s.getHeartbeatMessage()
+	for _, info := range hb.Replicas {
+		if info.ShardID == shardID {
+			for r := range info.Replicas {
+				if r == replicaID {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+func mustHaveReplica(t *testing.T,
+	s *store, shardID uint64, replicaID uint64) {
+	for i := 0; i < 100; i++ {
+		if hasReplica(s, shardID, replicaID) {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("failed to locate the replica")
+}