Repository: matrixorigin/matrixone
Pull request #3524 (2022-06-29 10:49:35)

add equal function for decimal128
**What type of PR is this?**

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #2565 

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/CODEOWNERS b/CODEOWNERS
new file mode 100644
index 000000000..9b8c18738
--- /dev/null
+++ b/CODEOWNERS
@@ -0,0 +1,1 @@
+/test/** @aressu1985
diff --git a/pkg/hakeeper/checkers/coordinator.go b/pkg/hakeeper/checkers/coordinator.go
index 8f70f9363..2ccf00681 100644
--- a/pkg/hakeeper/checkers/coordinator.go
+++ b/pkg/hakeeper/checkers/coordinator.go
@@ -17,13 +17,20 @@ package checkers
 import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/dnservice"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/logservice"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/syshealth"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
 	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
+// NB: Coordinator is assumed to be used in synchronous, single-threaded context.
 type Coordinator struct {
 	OperatorController *operator.Controller
+
+	// Considering the context of `Coordinator`,
+	// there is no need for a mutext to protect.
+	teardown    bool
+	teardownOps []*operator.Operator
 }
 
 func NewCoordinator() *Coordinator {
@@ -31,16 +38,29 @@ func NewCoordinator() *Coordinator {
 }
 
 func (c *Coordinator) Check(alloc util.IDAllocator, cluster hapb.ClusterInfo,
-	dn hapb.DNState, log hapb.LogState, currentTick uint64) []hapb.ScheduleCommand {
+	dnState hapb.DNState, logState hapb.LogState, currentTick uint64) []hapb.ScheduleCommand {
+
+	c.OperatorController.RemoveFinishedOperator(dnState, logState)
+
+	// if we've discovered unhealth already, no need to keep alive anymore.
+	if c.teardown {
+		return c.OperatorController.Dispatch(c.teardownOps, logState, dnState)
+	}
 
-	c.OperatorController.RemoveFinishedOperator(dn, log)
+	// check whether system health or not.
+	if operators, health := syshealth.Check(cluster, dnState, logState, currentTick); !health {
+		c.teardown = true
+		c.teardownOps = operators
+		return c.OperatorController.Dispatch(c.teardownOps, logState, dnState)
+	}
 
+	// system health, try to keep alive.
 	removing := c.OperatorController.GetRemovingReplicas()
 	adding := c.OperatorController.GetAddingReplicas()
 
 	operators := make([]*operator.Operator, 0)
-	operators = append(operators, logservice.Check(alloc, cluster, log, removing, adding, currentTick)...)
-	operators = append(operators, dnservice.Check(alloc, dn, currentTick)...)
+	operators = append(operators, logservice.Check(alloc, cluster, logState, removing, adding, currentTick)...)
+	operators = append(operators, dnservice.Check(alloc, dnState, currentTick)...)
 
-	return c.OperatorController.Dispatch(operators, log, dn)
+	return c.OperatorController.Dispatch(operators, logState, dnState)
 }
diff --git a/pkg/hakeeper/checkers/dnservice/README.md b/pkg/hakeeper/checkers/dnservice/README.md
deleted file mode 100644
index a577b752b..000000000
--- a/pkg/hakeeper/checkers/dnservice/README.md
+++ /dev/null
@@ -1,4 +0,0 @@
-## TODO
-
-- [x] Integration with OperatorController framework
-- [x] Replica ID generator
diff --git a/pkg/hakeeper/checkers/dnservice/check.go b/pkg/hakeeper/checkers/dnservice/check.go
index 6de06f19e..fdbf226db 100644
--- a/pkg/hakeeper/checkers/dnservice/check.go
+++ b/pkg/hakeeper/checkers/dnservice/check.go
@@ -16,7 +16,6 @@ package dnservice
 
 import (
 	"fmt"
-
 	"sort"
 
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
@@ -53,7 +52,7 @@ func Check(
 		steps := checkShard(shard, stores.WorkingStores(), idAlloc)
 		if len(steps) > 0 {
 			operators = append(operators,
-				operator.NewOperator("dn", shardID, operator.NoopEpoch, steps...),
+				operator.NewOperator("dnservice", shardID, operator.NoopEpoch, steps...),
 			)
 		}
 
diff --git a/pkg/hakeeper/checkers/syshealth/check.go b/pkg/hakeeper/checkers/syshealth/check.go
new file mode 100644
index 000000000..b3756e915
--- /dev/null
+++ b/pkg/hakeeper/checkers/syshealth/check.go
@@ -0,0 +1,227 @@
+// 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 syshealth
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
+)
+
+const (
+	defaultLogShardSize = 3
+)
+
+// Check checks system healthy or not.
+// If system wasn't healthy, we would generate
+// operators in order to shutdown all stores.
+func Check(
+	cluster hapb.ClusterInfo,
+	dnState hapb.DNState,
+	logState hapb.LogState,
+	currTick uint64,
+) ([]*operator.Operator, bool) {
+	sysHealthy := true
+
+	// parse all log stores for expired stores mainly
+	logStores := parseLogStores(logState, currTick)
+	if len(logStores.expired) == 0 {
+		return nil, sysHealthy
+	}
+
+	// check system healthy or not
+	for _, shard := range logShardsWithExpired(logStores.expired, logState, cluster) {
+		// if one of log shards wasn't healthy, the entire system wasn't healthy.
+		if !shard.healthy() {
+			sysHealthy = false
+			break
+		}
+	}
+
+	if sysHealthy {
+		return nil, sysHealthy
+	}
+
+	// parse all dn stores
+	dnStores := parseDnStores(dnState, currTick)
+
+	// generate operators to shutdown all stores
+	operators := make([]*operator.Operator, 0, logStores.length()+dnStores.length())
+	operators = append(operators, logStores.shutdownExpiredStores()...)
+	operators = append(operators, logStores.shutdownWorkingStores()...)
+	operators = append(operators, dnStores.shutdownExpiredStores()...)
+	operators = append(operators, dnStores.shutdownWorkingStores()...)
+
+	return operators, sysHealthy
+}
+
+// logShardMap is just an syntax sugar.
+type logShardMap map[uint64]*logShard
+
+func newLogShardMap() logShardMap {
+	return make(map[uint64]*logShard)
+}
+
+// registerExpiredReplica registers replica as expired.
+func (m logShardMap) registerExpiredReplica(replica logservice.LogReplicaInfo, cluster hapb.ClusterInfo) {
+	replicaID := replica.ReplicaID
+	shardID := replica.ShardID
+
+	if _, ok := m[shardID]; !ok {
+		m[shardID] = newLogShard(shardID, getLogShardSize(shardID, cluster))
+	}
+	m[shardID].registerExpiredReplica(replicaID)
+}
+
+// logShardsWithExpired gathers metadata for expired log shards.
+func logShardsWithExpired(
+	expiredStores map[util.StoreID]struct{},
+	logState hapb.LogState,
+	cluster hapb.ClusterInfo,
+) logShardMap {
+	expiredShardMap := newLogShardMap()
+	for id := range expiredStores {
+		expiredReplicas := logState.Stores[string(id)].Replicas
+		for _, replica := range expiredReplicas {
+			expiredShardMap.registerExpiredReplica(replica, cluster)
+		}
+	}
+	return expiredShardMap
+}
+
+// getLogShardSize gets raft group size for the specified shard.
+func getLogShardSize(shardID uint64, cluster hapb.ClusterInfo) uint64 {
+	for _, shardMeta := range cluster.LogShards {
+		if shardMeta.ShardID == shardID {
+			return shardMeta.NumberOfReplicas
+		}
+	}
+	return defaultLogShardSize
+}
+
+// logShard records metadata for log shard.
+type logShard struct {
+	shardID          uint64
+	numberOfReplicas uint64
+	expiredReplicas  map[uint64]struct{}
+}
+
+func newLogShard(shardID uint64, numberOfReplicas uint64) *logShard {
+	return &logShard{
+		shardID:          shardID,
+		numberOfReplicas: numberOfReplicas,
+		expiredReplicas:  make(map[uint64]struct{}),
+	}
+}
+
+// registerExpiredReplica registers expired replica ID.
+func (s *logShard) registerExpiredReplica(replicaID uint64) {
+	s.expiredReplicas[replicaID] = struct{}{}
+}
+
+// healthy checks whether log shard working or not.
+func (s *logShard) healthy() bool {
+	if s.numberOfReplicas > 0 &&
+		len(s.expiredReplicas)*2 >= int(s.numberOfReplicas) {
+		return false
+	}
+	return true
+}
+
+// storeSet separates stores as expired and working.
+type storeSet struct {
+	serviceType hapb.ServiceType
+	working     map[util.StoreID]struct{}
+	expired     map[util.StoreID]struct{}
+}
+
+func newStoreSet(serviceType hapb.ServiceType) *storeSet {
+	return &storeSet{
+		serviceType: serviceType,
+		working:     make(map[util.StoreID]struct{}),
+		expired:     make(map[util.StoreID]struct{}),
+	}
+}
+
+// length returns number of all stores within this set.
+func (s *storeSet) length() int {
+	return len(s.working) + len(s.expired)
+}
+
+// shutdownExpiredStores
+func (s *storeSet) shutdownExpiredStores() []*operator.Operator {
+	return shutdownStores(s.serviceType, s.expired)
+}
+
+// shutdownWorkingStores generates operators to shutdown working stores.
+func (s *storeSet) shutdownWorkingStores() []*operator.Operator {
+	return shutdownStores(s.serviceType, s.working)
+}
+
+// parseLogStores separates log stores as expired and working.
+func parseLogStores(logState hapb.LogState, currTick uint64) *storeSet {
+	set := newStoreSet(hapb.LogService)
+	for id, storeInfo := range logState.Stores {
+		if hakeeper.ExpiredTick(storeInfo.Tick, hakeeper.LogStoreTimeout) < currTick {
+			set.expired[util.StoreID(id)] = struct{}{}
+		} else {
+			set.working[util.StoreID(id)] = struct{}{}
+		}
+	}
+	return set
+}
+
+// parseDnStores separates dn stores as expired and working.
+func parseDnStores(dnState hapb.DNState, currTick uint64) *storeSet {
+	set := newStoreSet(hapb.DnService)
+	for id, storeInfo := range dnState.Stores {
+		if hakeeper.ExpiredTick(storeInfo.Tick, hakeeper.DnStoreTimeout) < currTick {
+			set.expired[util.StoreID(id)] = struct{}{}
+		} else {
+			set.working[util.StoreID(id)] = struct{}{}
+		}
+	}
+	return set
+}
+
+// shutdownStores generates operators to shutdown stores.
+func shutdownStores(serviceType hapb.ServiceType, stores map[util.StoreID]struct{}) []*operator.Operator {
+	ops := make([]*operator.Operator, 0, len(stores))
+
+	switch serviceType {
+	case hapb.LogService:
+		for id := range stores {
+			op := operator.NewOperator(
+				"logservice", operator.NoopShardID, operator.NoopEpoch,
+				operator.StopLogStore{StoreID: string(id)},
+			)
+			ops = append(ops, op)
+		}
+	case hapb.DnService:
+		for id := range stores {
+			op := operator.NewOperator(
+				"dnservice", operator.NoopShardID, operator.NoopEpoch,
+				operator.StopDnStore{StoreID: string(id)},
+			)
+			ops = append(ops, op)
+		}
+	default:
+		panic("unexpected service type")
+	}
+
+	return ops
+}
diff --git a/pkg/hakeeper/operator/controller.go b/pkg/hakeeper/operator/controller.go
index 7e4af0bba..05d9e9e13 100644
--- a/pkg/hakeeper/operator/controller.go
+++ b/pkg/hakeeper/operator/controller.go
@@ -21,8 +21,9 @@
 package operator
 
 import (
-	"github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	"sync"
+
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 )
 
 // Controller is used to manage operators.
@@ -97,7 +98,7 @@ func (c *Controller) GetAddingReplicas() (adding map[uint64][]uint64) {
 	return
 }
 
-func (c *Controller) RemoveFinishedOperator(dnState hakeeper.DNState, state hakeeper.LogState) {
+func (c *Controller) RemoveFinishedOperator(dnState hapb.DNState, state hapb.LogState) {
 	for _, ops := range c.operators {
 		for _, op := range ops {
 			op.Check(state, dnState)
@@ -109,91 +110,108 @@ func (c *Controller) RemoveFinishedOperator(dnState hakeeper.DNState, state hake
 	}
 }
 
-func (c *Controller) Dispatch(ops []*Operator, logState hakeeper.LogState, dnState hakeeper.DNState) (commands []hakeeper.ScheduleCommand) {
+func (c *Controller) Dispatch(ops []*Operator, logState hapb.LogState, dnState hapb.DNState) (commands []hapb.ScheduleCommand) {
 	for _, op := range ops {
 		c.operators[op.shardID] = append(c.operators[op.shardID], op)
 		step := op.Check(logState, dnState)
-		var cmd hakeeper.ScheduleCommand
+		var cmd hapb.ScheduleCommand
 		switch st := step.(type) {
 		case AddLogService:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.Target,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:      st.StoreID,
 						ShardID:   st.ShardID,
 						ReplicaID: st.ReplicaID,
 						Epoch:     st.Epoch,
 					},
-					ChangeType: hakeeper.AddReplica,
+					ChangeType: hapb.AddReplica,
 				},
-				ServiceType: hakeeper.LogService,
+				ServiceType: hapb.LogService,
 			}
 		case RemoveLogService:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.Target,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:      st.StoreID,
 						ShardID:   st.ShardID,
 						ReplicaID: st.ReplicaID,
 					},
-					ChangeType: hakeeper.RemoveReplica,
+					ChangeType: hapb.RemoveReplica,
 				},
-				ServiceType: hakeeper.LogService,
+				ServiceType: hapb.LogService,
 			}
 		case StartLogService:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.StoreID,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:      st.StoreID,
 						ShardID:   st.ShardID,
 						ReplicaID: st.ReplicaID,
 					},
-					ChangeType: hakeeper.StartReplica,
+					ChangeType: hapb.StartReplica,
 				},
-				ServiceType: hakeeper.LogService,
+				ServiceType: hapb.LogService,
 			}
 		case StopLogService:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.StoreID,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:    st.StoreID,
 						ShardID: st.ShardID,
 					},
-					ChangeType: hakeeper.StopReplica,
+					ChangeType: hapb.StopReplica,
 				},
-				ServiceType: hakeeper.LogService,
+				ServiceType: hapb.LogService,
 			}
 		case AddDnReplica:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.StoreID,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:      st.StoreID,
 						ShardID:   st.ShardID,
 						ReplicaID: st.ReplicaID,
 					},
-					ChangeType: hakeeper.AddReplica,
+					ChangeType: hapb.AddReplica,
 				},
-				ServiceType: hakeeper.DnService,
+				ServiceType: hapb.DnService,
 			}
 		case RemoveDnReplica:
-			cmd = hakeeper.ScheduleCommand{
+			cmd = hapb.ScheduleCommand{
 				UUID: st.StoreID,
-				ConfigChange: hakeeper.ConfigChange{
-					Replica: hakeeper.Replica{
+				ConfigChange: &hapb.ConfigChange{
+					Replica: hapb.Replica{
 						UUID:      st.StoreID,
 						ShardID:   st.ShardID,
 						ReplicaID: st.ReplicaID,
 					},
-					ChangeType: hakeeper.RemoveReplica,
+					ChangeType: hapb.RemoveReplica,
+				},
+				ServiceType: hapb.DnService,
+			}
+		case StopDnStore:
+			cmd = hapb.ScheduleCommand{
+				UUID: st.StoreID,
+				ShutdownStore: &hapb.ShutdownStore{
+					StoreID: st.StoreID,
 				},
-				ServiceType: hakeeper.DnService,
+				ServiceType: hapb.DnService,
+			}
+		case StopLogStore:
+			cmd = hapb.ScheduleCommand{
+				UUID: st.StoreID,
+				ShutdownStore: &hapb.ShutdownStore{
+					StoreID: st.StoreID,
+				},
+				ServiceType: hapb.LogService,
 			}
 		}
+
 		commands = append(commands, cmd)
 	}
 
diff --git a/pkg/hakeeper/operator/operator.go b/pkg/hakeeper/operator/operator.go
index 2caf7fe96..ef344ba00 100644
--- a/pkg/hakeeper/operator/operator.go
+++ b/pkg/hakeeper/operator/operator.go
@@ -28,8 +28,9 @@ import (
 )
 
 const (
-	ExpireTime        = 5 * time.Minute
-	NoopEpoch  uint64 = 0
+	ExpireTime         = 5 * time.Minute
+	NoopEpoch   uint64 = 0
+	NoopShardID uint64 = 0
 )
 
 // Operator contains execution steps generated by scheduler.
diff --git a/pkg/hakeeper/operator/steps.go b/pkg/hakeeper/operator/steps.go
index b8ada82e7..87adc0e68 100644
--- a/pkg/hakeeper/operator/steps.go
+++ b/pkg/hakeeper/operator/steps.go
@@ -153,3 +153,35 @@ func (a RemoveDnReplica) IsFinish(_ hakeeper.LogState, state hakeeper.DNState) b
 	}
 	return true
 }
+
+// StopDnStore corresponds to dn store shutdown command.
+type StopDnStore struct {
+	StoreID string
+}
+
+func (a StopDnStore) String() string {
+	return fmt.Sprintf("stopping dn store %s", a.StoreID)
+}
+
+func (a StopDnStore) IsFinish(_ hakeeper.LogState, state hakeeper.DNState) bool {
+	if _, ok := state.Stores[a.StoreID]; ok {
+		return false
+	}
+	return true
+}
+
+// StopLogStore corresponds to log store shutdown command.
+type StopLogStore struct {
+	StoreID string
+}
+
+func (a StopLogStore) String() string {
+	return fmt.Sprintf("stopping log store %s", a.StoreID)
+}
+
+func (a StopLogStore) IsFinish(state hakeeper.LogState, _ hakeeper.DNState) bool {
+	if _, ok := state.Stores[a.StoreID]; ok {
+		return false
+	}
+	return true
+}
diff --git a/pkg/logservice/commands.go b/pkg/logservice/commands.go
index b2ff4b51e..9dfe068dd 100644
--- a/pkg/logservice/commands.go
+++ b/pkg/logservice/commands.go
@@ -21,17 +21,24 @@ import (
 
 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")
+		if cmd.GetConfigChange() != nil {
+			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")
+			}
+			return
+		}
+
+		if cmd.GetShutdownStore() != nil {
+			// FIXME: add logic to shutdown store
 		}
 	}
 }
diff --git a/pkg/pb/hakeeper/hakeeper.pb.go b/pkg/pb/hakeeper/hakeeper.pb.go
index a09351b5c..9987252f9 100644
--- a/pkg/pb/hakeeper/hakeeper.pb.go
+++ b/pkg/pb/hakeeper/hakeeper.pb.go
@@ -5,13 +5,14 @@ package hakeeper
 
 import (
 	fmt "fmt"
+	io "io"
+	math "math"
+	math_bits "math/bits"
+
 	_ "github.com/gogo/protobuf/gogoproto"
 	proto "github.com/gogo/protobuf/proto"
 	logservice "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	metadata "github.com/matrixorigin/matrixone/pkg/pb/metadata"
-	io "io"
-	math "math"
-	math_bits "math/bits"
 )
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -212,22 +213,71 @@ func (m *ConfigChange) GetChangeType() ConfigChangeType {
 	return AddReplica
 }
 
+// ShutdownStore would stop store.
+type ShutdownStore struct {
+	StoreID              string   `protobuf:"bytes,1,opt,name=StoreID,proto3" json:"StoreID,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ShutdownStore) Reset()         { *m = ShutdownStore{} }
+func (m *ShutdownStore) String() string { return proto.CompactTextString(m) }
+func (*ShutdownStore) ProtoMessage()    {}
+func (*ShutdownStore) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5e1506f3aa5330eb, []int{2}
+}
+func (m *ShutdownStore) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *ShutdownStore) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_ShutdownStore.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *ShutdownStore) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ShutdownStore.Merge(m, src)
+}
+func (m *ShutdownStore) XXX_Size() int {
+	return m.Size()
+}
+func (m *ShutdownStore) XXX_DiscardUnknown() {
+	xxx_messageInfo_ShutdownStore.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ShutdownStore proto.InternalMessageInfo
+
+func (m *ShutdownStore) GetStoreID() string {
+	if m != nil {
+		return m.StoreID
+	}
+	return ""
+}
+
 // ScheduleCommand contains a shard schedule command.
 type ScheduleCommand struct {
 	// UUID which store the ScheduleCommand is sent to
-	UUID                 string       `protobuf:"bytes,1,opt,name=UUID,proto3" json:"UUID,omitempty"`
-	ConfigChange         ConfigChange `protobuf:"bytes,2,opt,name=ConfigChange,proto3" json:"ConfigChange"`
-	ServiceType          ServiceType  `protobuf:"varint,3,opt,name=ServiceType,proto3,enum=hakeeper.ServiceType" json:"ServiceType,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}     `json:"-"`
-	XXX_unrecognized     []byte       `json:"-"`
-	XXX_sizecache        int32        `json:"-"`
+	UUID                 string         `protobuf:"bytes,1,opt,name=UUID,proto3" json:"UUID,omitempty"`
+	ConfigChange         *ConfigChange  `protobuf:"bytes,2,opt,name=ConfigChange,proto3" json:"ConfigChange,omitempty"`
+	ServiceType          ServiceType    `protobuf:"varint,3,opt,name=ServiceType,proto3,enum=hakeeper.ServiceType" json:"ServiceType,omitempty"`
+	ShutdownStore        *ShutdownStore `protobuf:"bytes,4,opt,name=ShutdownStore,proto3" json:"ShutdownStore,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}       `json:"-"`
+	XXX_unrecognized     []byte         `json:"-"`
+	XXX_sizecache        int32          `json:"-"`
 }
 
 func (m *ScheduleCommand) Reset()         { *m = ScheduleCommand{} }
 func (m *ScheduleCommand) String() string { return proto.CompactTextString(m) }
 func (*ScheduleCommand) ProtoMessage()    {}
 func (*ScheduleCommand) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{2}
+	return fileDescriptor_5e1506f3aa5330eb, []int{3}
 }
 func (m *ScheduleCommand) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -263,11 +313,11 @@ func (m *ScheduleCommand) GetUUID() string {
 	return ""
 }
 
-func (m *ScheduleCommand) GetConfigChange() ConfigChange {
+func (m *ScheduleCommand) GetConfigChange() *ConfigChange {
 	if m != nil {
 		return m.ConfigChange
 	}
-	return ConfigChange{}
+	return nil
 }
 
 func (m *ScheduleCommand) GetServiceType() ServiceType {
@@ -277,6 +327,13 @@ func (m *ScheduleCommand) GetServiceType() ServiceType {
 	return LogService
 }
 
+func (m *ScheduleCommand) GetShutdownStore() *ShutdownStore {
+	if m != nil {
+		return m.ShutdownStore
+	}
+	return nil
+}
+
 type CommandBatch struct {
 	Term                 uint64            `protobuf:"varint,1,opt,name=Term,proto3" json:"Term,omitempty"`
 	Commands             []ScheduleCommand `protobuf:"bytes,2,rep,name=Commands,proto3" json:"Commands"`
@@ -289,7 +346,7 @@ func (m *CommandBatch) Reset()         { *m = CommandBatch{} }
 func (m *CommandBatch) String() string { return proto.CompactTextString(m) }
 func (*CommandBatch) ProtoMessage()    {}
 func (*CommandBatch) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{3}
+	return fileDescriptor_5e1506f3aa5330eb, []int{4}
 }
 func (m *CommandBatch) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -345,7 +402,7 @@ func (m *DNStoreInfo) Reset()         { *m = DNStoreInfo{} }
 func (m *DNStoreInfo) String() string { return proto.CompactTextString(m) }
 func (*DNStoreInfo) ProtoMessage()    {}
 func (*DNStoreInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{4}
+	return fileDescriptor_5e1506f3aa5330eb, []int{5}
 }
 func (m *DNStoreInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -402,7 +459,7 @@ func (m *DNState) Reset()         { *m = DNState{} }
 func (m *DNState) String() string { return proto.CompactTextString(m) }
 func (*DNState) ProtoMessage()    {}
 func (*DNState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{5}
+	return fileDescriptor_5e1506f3aa5330eb, []int{6}
 }
 func (m *DNState) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -453,7 +510,7 @@ func (m *ClusterInfo) Reset()         { *m = ClusterInfo{} }
 func (m *ClusterInfo) String() string { return proto.CompactTextString(m) }
 func (*ClusterInfo) ProtoMessage()    {}
 func (*ClusterInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{6}
+	return fileDescriptor_5e1506f3aa5330eb, []int{7}
 }
 func (m *ClusterInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -512,7 +569,7 @@ func (m *LogStoreInfo) Reset()         { *m = LogStoreInfo{} }
 func (m *LogStoreInfo) String() string { return proto.CompactTextString(m) }
 func (*LogStoreInfo) ProtoMessage()    {}
 func (*LogStoreInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{7}
+	return fileDescriptor_5e1506f3aa5330eb, []int{8}
 }
 func (m *LogStoreInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -582,8 +639,9 @@ type LogState struct {
 	// different replicas and thus reflect a more accurate description on each
 	// shard.
 	Shards map[uint64]logservice.LogShardInfo `protobuf:"bytes,1,rep,name=Shards,proto3" json:"Shards" protobuf_key:"varint,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
-	// Stores is keyed by log store UUID, it contains details found on each store.
-	// Each LogStoreInfo here reflects what was last reported by each Log store.
+	// Stores is keyed by log store UUID, it contains details found on each
+	// store. Each LogStoreInfo here reflects what was last reported by each Log
+	// store.
 	Stores               map[string]LogStoreInfo `protobuf:"bytes,2,rep,name=Stores,proto3" json:"Stores" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
 	XXX_NoUnkeyedLiteral struct{}                `json:"-"`
 	XXX_unrecognized     []byte                  `json:"-"`
@@ -594,7 +652,7 @@ func (m *LogState) Reset()         { *m = LogState{} }
 func (m *LogState) String() string { return proto.CompactTextString(m) }
 func (*LogState) ProtoMessage()    {}
 func (*LogState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{8}
+	return fileDescriptor_5e1506f3aa5330eb, []int{9}
 }
 func (m *LogState) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -653,7 +711,7 @@ func (m *HAKeeperState) Reset()         { *m = HAKeeperState{} }
 func (m *HAKeeperState) String() string { return proto.CompactTextString(m) }
 func (*HAKeeperState) ProtoMessage()    {}
 func (*HAKeeperState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{9}
+	return fileDescriptor_5e1506f3aa5330eb, []int{10}
 }
 func (m *HAKeeperState) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -729,7 +787,7 @@ func (m *RSMState) Reset()         { *m = RSMState{} }
 func (m *RSMState) String() string { return proto.CompactTextString(m) }
 func (*RSMState) ProtoMessage()    {}
 func (*RSMState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5e1506f3aa5330eb, []int{10}
+	return fileDescriptor_5e1506f3aa5330eb, []int{11}
 }
 func (m *RSMState) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -819,6 +877,7 @@ func init() {
 	proto.RegisterEnum("hakeeper.ServiceType", ServiceType_name, ServiceType_value)
 	proto.RegisterType((*Replica)(nil), "hakeeper.Replica")
 	proto.RegisterType((*ConfigChange)(nil), "hakeeper.ConfigChange")
+	proto.RegisterType((*ShutdownStore)(nil), "hakeeper.ShutdownStore")
 	proto.RegisterType((*ScheduleCommand)(nil), "hakeeper.ScheduleCommand")
 	proto.RegisterType((*CommandBatch)(nil), "hakeeper.CommandBatch")
 	proto.RegisterType((*DNStoreInfo)(nil), "hakeeper.DNStoreInfo")
@@ -838,63 +897,65 @@ func init() {
 func init() { proto.RegisterFile("hakeeper.proto", fileDescriptor_5e1506f3aa5330eb) }
 
 var fileDescriptor_5e1506f3aa5330eb = []byte{
-	// 897 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x56, 0xdd, 0x8e, 0xdb, 0x44,
-	0x14, 0x5e, 0x27, 0xde, 0x6c, 0x72, 0xbc, 0x49, 0xdd, 0x51, 0x77, 0x6b, 0x22, 0x08, 0xc1, 0x42,
-	0x28, 0x2a, 0xc5, 0x55, 0x03, 0x08, 0x68, 0x83, 0x60, 0x37, 0xa9, 0x20, 0xa2, 0xac, 0xc0, 0x69,
-	0xa5, 0x0a, 0x2e, 0x90, 0xd7, 0x9e, 0x38, 0x51, 0x7e, 0x26, 0xb2, 0x9d, 0x15, 0x7b, 0xc1, 0x2d,
-	0xcf, 0xc0, 0x25, 0xd7, 0x5c, 0xf0, 0x1c, 0x95, 0xb8, 0xd9, 0x27, 0x40, 0xb0, 0x4f, 0x82, 0x7c,
-	0x3c, 0x63, 0xcf, 0x64, 0xb3, 0xc0, 0xf6, 0xce, 0xe7, 0xef, 0x9b, 0xf3, 0xf3, 0x9d, 0x19, 0x43,
-	0x63, 0xe2, 0xcd, 0x28, 0x5d, 0xd1, 0xc8, 0x59, 0x45, 0x2c, 0x61, 0xa4, 0x2a, 0xe4, 0xe6, 0x7b,
-	0xe1, 0x34, 0x99, 0xac, 0x4f, 0x1d, 0x9f, 0x2d, 0x1e, 0x84, 0x2c, 0x64, 0x0f, 0xd0, 0xe1, 0x74,
-	0x3d, 0x46, 0x09, 0x05, 0xfc, 0xca, 0x02, 0x9b, 0xe6, 0x9c, 0x85, 0x31, 0x8d, 0xce, 0xa6, 0x3e,
-	0xe5, 0x9a, 0xc6, 0x82, 0x26, 0x5e, 0xe0, 0x25, 0x5e, 0x26, 0xdb, 0x33, 0xd8, 0x73, 0xe9, 0x6a,
-	0x3e, 0xf5, 0x3d, 0x42, 0x40, 0x7f, 0xfe, 0x7c, 0x38, 0xb0, 0xb4, 0xb6, 0xd6, 0xa9, 0xb9, 0xf8,
-	0x4d, 0x2c, 0xd8, 0x1b, 0x4d, 0xbc, 0x28, 0x18, 0x0e, 0xac, 0x52, 0x5b, 0xeb, 0xe8, 0xae, 0x10,
-	0xc9, 0xeb, 0x50, 0xe3, 0x81, 0xc3, 0x81, 0x55, 0x46, 0x5b, 0xa1, 0x20, 0x77, 0x60, 0xf7, 0xc9,
-	0x8a, 0xf9, 0x13, 0x4b, 0x47, 0x4b, 0x26, 0xd8, 0x3f, 0xc1, 0x7e, 0x9f, 0x2d, 0xc7, 0xd3, 0xb0,
-	0x3f, 0xf1, 0x96, 0x21, 0x25, 0x0f, 0xf3, 0xc3, 0xf1, 0x50, 0xa3, 0x7b, 0xdb, 0xc9, 0x2b, 0xe7,
-	0x86, 0x63, 0xfd, 0xe5, 0x9f, 0x6f, 0xee, 0xb8, 0x79, 0x92, 0x8f, 0x00, 0xb2, 0xe0, 0x67, 0xe7,
-	0x2b, 0x8a, 0x39, 0x35, 0xba, 0xcd, 0x22, 0x4a, 0x86, 0x4f, 0x3d, 0x5c, 0xc9, 0xdb, 0xfe, 0x55,
-	0x83, 0x5b, 0x23, 0x7f, 0x42, 0x83, 0xf5, 0x9c, 0xf6, 0xd9, 0x62, 0xe1, 0x2d, 0x83, 0xad, 0x45,
-	0x7f, 0xae, 0xa6, 0x89, 0xa7, 0x18, 0xdd, 0xc3, 0xed, 0xa7, 0xf0, 0x04, 0xd5, 0xc2, 0x3e, 0x02,
-	0x63, 0x94, 0xb5, 0x1d, 0xd3, 0x2c, 0x63, 0x9a, 0x07, 0x05, 0x80, 0x64, 0x74, 0x65, 0x4f, 0xfb,
-	0x87, 0xf4, 0x68, 0xcc, 0xec, 0xd8, 0x4b, 0xfc, 0x49, 0x9a, 0xde, 0x33, 0x1a, 0x2d, 0x30, 0x3d,
-	0xdd, 0xc5, 0x6f, 0xf2, 0x18, 0xaa, 0xdc, 0x27, 0xb6, 0x4a, 0xed, 0x72, 0xc7, 0xe8, 0xbe, 0x26,
-	0x21, 0xab, 0xf5, 0xf1, 0xec, 0xf2, 0x00, 0xfb, 0x05, 0x18, 0x83, 0x93, 0x51, 0xc2, 0x22, 0x3a,
-	0x5c, 0x8e, 0x19, 0xe2, 0x4f, 0xfd, 0x59, 0x8e, 0x3f, 0xf5, 0x67, 0xe4, 0x43, 0xa8, 0xe0, 0x90,
-	0x05, 0xfa, 0x5d, 0x47, 0x62, 0xd1, 0xe0, 0x24, 0x23, 0xc0, 0x72, 0xcc, 0x38, 0x36, 0x77, 0xb6,
-	0x7f, 0xd1, 0x60, 0x2f, 0x85, 0xf6, 0x12, 0x4a, 0x1e, 0x43, 0x05, 0xcf, 0x88, 0x2d, 0x0d, 0x21,
-	0xde, 0x28, 0x12, 0xe4, 0x2e, 0x4e, 0x66, 0x7f, 0xb2, 0x4c, 0xa2, 0xf3, 0x1c, 0x08, 0x55, 0xcd,
-	0x6f, 0xc0, 0x90, 0x8c, 0xc4, 0x84, 0xf2, 0x8c, 0x9e, 0xf3, 0x01, 0xa5, 0x9f, 0xe4, 0x5d, 0xd8,
-	0x3d, 0xf3, 0xe6, 0x6b, 0x31, 0x98, 0x03, 0x15, 0x9c, 0x97, 0xe6, 0x66, 0x3e, 0x8f, 0x4a, 0x1f,
-	0x6b, 0xf6, 0xcf, 0x1a, 0x18, 0xfd, 0xf9, 0x3a, 0x4e, 0x68, 0x84, 0x55, 0x7f, 0x02, 0x55, 0x5e,
-	0x87, 0x48, 0xf0, 0xae, 0x93, 0xef, 0x05, 0xb7, 0xb8, 0xd4, 0x67, 0x51, 0xde, 0x3f, 0xe1, 0x4e,
-	0x7a, 0x50, 0x7b, 0xca, 0x42, 0xa5, 0x3f, 0x56, 0x11, 0x2b, 0x4c, 0x4a, 0x70, 0x11, 0x60, 0x5f,
-	0x68, 0xb0, 0x9f, 0x4a, 0xff, 0xda, 0xff, 0x36, 0x18, 0xae, 0x37, 0x4e, 0x8e, 0x82, 0x20, 0xa2,
-	0x71, 0x8c, 0x45, 0xd6, 0x5c, 0x59, 0x45, 0xde, 0x81, 0x06, 0x27, 0x8d, 0x70, 0x2a, 0xa3, 0xd3,
-	0x86, 0x96, 0xbc, 0x0d, 0xf5, 0x2f, 0x58, 0x1c, 0x4f, 0x57, 0xc2, 0x4d, 0x47, 0x37, 0x55, 0x49,
-	0x7a, 0x50, 0xe5, 0xdb, 0x15, 0x5b, 0xbb, 0x58, 0x51, 0x53, 0x9e, 0xf8, 0x53, 0x16, 0x8a, 0xbd,
-	0x2e, 0x86, 0x9e, 0x47, 0xd8, 0xbf, 0x97, 0xa0, 0x8a, 0x25, 0xa5, 0x73, 0xef, 0xe5, 0xd4, 0xc9,
-	0xda, 0xda, 0x2a, 0x46, 0x23, 0x7c, 0x9c, 0xcc, 0x41, 0x1d, 0xbc, 0xe8, 0xad, 0x60, 0x4d, 0xe9,
-	0xfa, 0xe8, 0x6b, 0x69, 0x33, 0x02, 0x43, 0x82, 0x96, 0x69, 0xa3, 0x67, 0xb4, 0x71, 0x54, 0xda,
-	0x58, 0x1b, 0x45, 0xe6, 0xbc, 0x96, 0x98, 0xd3, 0xfc, 0xf6, 0xbf, 0xb8, 0x78, 0x5f, 0x05, 0x3d,
-	0xdc, 0x48, 0x79, 0x0b, 0x19, 0xff, 0xd0, 0xa0, 0xfe, 0xe5, 0xd1, 0x57, 0xe8, 0x94, 0x75, 0x6d,
-	0x1b, 0x09, 0x3e, 0x55, 0x18, 0x7b, 0x95, 0xe9, 0x92, 0x91, 0xf7, 0x41, 0x61, 0xf8, 0xc3, 0x7c,
-	0x17, 0x91, 0x1a, 0xca, 0xcd, 0xca, 0x0d, 0xe2, 0x66, 0x15, 0x3b, 0xfb, 0x41, 0x31, 0x47, 0xe4,
-	0x89, 0xd1, 0x25, 0x57, 0xfb, 0x2f, 0xc6, 0x2f, 0x64, 0xfb, 0x37, 0x1d, 0xaa, 0xee, 0xe8, 0xeb,
-	0xeb, 0x0b, 0x39, 0x84, 0xca, 0x09, 0xfd, 0x31, 0xc9, 0x1f, 0x10, 0x2e, 0xe5, 0x37, 0x5b, 0x59,
-	0xba, 0xd9, 0xbe, 0x03, 0x73, 0xe3, 0xfe, 0x4a, 0x29, 0x9b, 0x52, 0xa1, 0x23, 0x3d, 0x0c, 0xfc,
-	0xb4, 0xcd, 0xab, 0x4e, 0x21, 0xc5, 0x15, 0x1c, 0xf2, 0x99, 0xbc, 0xb8, 0x19, 0xcd, 0xdf, 0xda,
-	0x02, 0x9a, 0xfb, 0x20, 0x9a, 0xb4, 0xbb, 0x72, 0x4b, 0x2b, 0xaf, 0xd0, 0xd2, 0xbd, 0xff, 0xdb,
-	0xd2, 0xcd, 0xd1, 0x57, 0x6f, 0x36, 0xfa, 0xe6, 0xf7, 0x70, 0xb0, 0xb5, 0x33, 0x37, 0x22, 0xaf,
-	0xfc, 0x08, 0xc9, 0xfb, 0xd0, 0x83, 0x86, 0xda, 0xa1, 0x2d, 0xa8, 0x77, 0x64, 0x54, 0x5d, 0x8a,
-	0xbe, 0xf7, 0x02, 0xcc, 0xcd, 0x07, 0x9a, 0x34, 0x00, 0x8e, 0x82, 0x80, 0x5f, 0x27, 0xe6, 0x0e,
-	0xb9, 0x0d, 0x75, 0x97, 0x2e, 0xd8, 0x19, 0x15, 0x2a, 0x8d, 0x98, 0xb0, 0x3f, 0x4a, 0xbc, 0x28,
-	0x11, 0x9a, 0x12, 0xb9, 0x85, 0x6b, 0xb9, 0x12, 0x8a, 0xf2, 0xbd, 0xfb, 0xca, 0x83, 0x9b, 0x82,
-	0xa6, 0x69, 0x66, 0x1a, 0x73, 0x87, 0xd4, 0xa1, 0x36, 0x58, 0x0a, 0x51, 0x3b, 0x36, 0x2f, 0xfe,
-	0x6e, 0xed, 0xbc, 0xbc, 0x6c, 0x69, 0x17, 0x97, 0x2d, 0xed, 0xaf, 0xcb, 0x96, 0x76, 0x5a, 0xc1,
-	0xbf, 0xa1, 0xf7, 0xff, 0x09, 0x00, 0x00, 0xff, 0xff, 0xc9, 0xe8, 0x88, 0x09, 0x7a, 0x09, 0x00,
+	// 929 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x56, 0x4f, 0x6f, 0xe3, 0x44,
+	0x14, 0xaf, 0x13, 0x37, 0x4d, 0x9e, 0x9b, 0x6c, 0x76, 0xb4, 0x6d, 0x4d, 0x04, 0xa1, 0x58, 0x08,
+	0x95, 0x65, 0xc9, 0x6a, 0x03, 0x08, 0xe8, 0x76, 0x85, 0xda, 0x66, 0x05, 0x15, 0x4b, 0x05, 0xce,
+	0xae, 0xb4, 0x82, 0x03, 0x72, 0xed, 0x49, 0x1c, 0x25, 0xf1, 0x44, 0xb6, 0x53, 0xe8, 0x81, 0x2b,
+	0x9f, 0x81, 0xcf, 0xc0, 0x81, 0xcf, 0xb1, 0x12, 0x97, 0x5e, 0xb9, 0x20, 0xe8, 0x27, 0x41, 0x7e,
+	0x9e, 0x19, 0xcf, 0xb8, 0x29, 0xb0, 0x7b, 0xf3, 0xfb, 0xf7, 0xf3, 0xfc, 0xde, 0xfb, 0xcd, 0xb3,
+	0xa1, 0x15, 0x7a, 0x53, 0x4a, 0x17, 0x34, 0xee, 0x2d, 0x62, 0x96, 0x32, 0x52, 0x17, 0x76, 0xe7,
+	0xfd, 0xf1, 0x24, 0x0d, 0x97, 0x67, 0x3d, 0x9f, 0xcd, 0xef, 0x8f, 0xd9, 0x98, 0xdd, 0xc7, 0x84,
+	0xb3, 0xe5, 0x08, 0x2d, 0x34, 0xf0, 0x29, 0x2f, 0xec, 0xb4, 0x67, 0x6c, 0x9c, 0xd0, 0xf8, 0x7c,
+	0xe2, 0x53, 0xee, 0x69, 0xcd, 0x69, 0xea, 0x05, 0x5e, 0xea, 0xe5, 0xb6, 0x33, 0x85, 0x0d, 0x97,
+	0x2e, 0x66, 0x13, 0xdf, 0x23, 0x04, 0xcc, 0x67, 0xcf, 0x4e, 0x06, 0xb6, 0xb1, 0x6b, 0xec, 0x35,
+	0x5c, 0x7c, 0x26, 0x36, 0x6c, 0x0c, 0x43, 0x2f, 0x0e, 0x4e, 0x06, 0x76, 0x65, 0xd7, 0xd8, 0x33,
+	0x5d, 0x61, 0x92, 0xd7, 0xa1, 0xc1, 0x0b, 0x4f, 0x06, 0x76, 0x15, 0x63, 0x85, 0x83, 0xdc, 0x81,
+	0xf5, 0xc7, 0x0b, 0xe6, 0x87, 0xb6, 0x89, 0x91, 0xdc, 0x70, 0x7e, 0x82, 0xcd, 0x63, 0x16, 0x8d,
+	0x26, 0xe3, 0xe3, 0xd0, 0x8b, 0xc6, 0x94, 0x3c, 0x90, 0x2f, 0xc7, 0x97, 0x5a, 0xfd, 0xdb, 0x3d,
+	0xc9, 0x9c, 0x07, 0x8e, 0xcc, 0x17, 0x7f, 0xbe, 0xb9, 0xe6, 0xca, 0x43, 0xee, 0x03, 0xe4, 0xc5,
+	0x4f, 0x2f, 0x16, 0x14, 0xcf, 0xd4, 0xea, 0x77, 0x8a, 0x2a, 0x15, 0x3e, 0xcb, 0x70, 0x95, 0x6c,
+	0xe7, 0x5d, 0x68, 0x0e, 0xc3, 0x65, 0x1a, 0xb0, 0x1f, 0xa2, 0x61, 0xca, 0x62, 0x8a, 0xec, 0xb2,
+	0x07, 0x49, 0x5a, 0x98, 0xce, 0x1f, 0x06, 0xdc, 0x1a, 0xfa, 0x21, 0x0d, 0x96, 0x33, 0x7a, 0xcc,
+	0xe6, 0x73, 0x2f, 0x0a, 0x56, 0xf6, 0x67, 0x5f, 0x67, 0x84, 0x07, 0xb2, 0xfa, 0xdb, 0xab, 0x0f,
+	0xe4, 0xea, 0xec, 0x3f, 0x06, 0x6b, 0x98, 0xcf, 0x06, 0xb9, 0x54, 0x91, 0xcb, 0x56, 0x51, 0xaa,
+	0x04, 0x5d, 0x35, 0x93, 0x3c, 0x2a, 0xf1, 0xc0, 0x26, 0x5b, 0xfd, 0x1d, 0xa5, 0x54, 0x0d, 0xbb,
+	0x7a, 0xb6, 0xf3, 0x7d, 0x76, 0x66, 0xa4, 0x74, 0xe4, 0xa5, 0x7e, 0x98, 0xf1, 0x7a, 0x4a, 0xe3,
+	0x39, 0xf2, 0x32, 0x5d, 0x7c, 0x26, 0x0f, 0xa1, 0xce, 0x73, 0x12, 0xbb, 0xb2, 0x5b, 0xdd, 0xb3,
+	0xfa, 0xaf, 0x29, 0xe8, 0x7a, 0x63, 0xf8, 0x88, 0x64, 0x81, 0xf3, 0x1c, 0xac, 0xc1, 0x69, 0xde,
+	0xc9, 0x68, 0xc4, 0x10, 0x7f, 0xe2, 0x4f, 0x25, 0xfe, 0xc4, 0x9f, 0x92, 0x8f, 0xa0, 0x86, 0x42,
+	0x12, 0xe8, 0x3b, 0x3d, 0x45, 0xa9, 0x83, 0xd3, 0x5c, 0x64, 0xd1, 0x88, 0x71, 0x6c, 0x9e, 0xec,
+	0xfc, 0x62, 0xc0, 0x46, 0x06, 0xed, 0xa5, 0x94, 0x3c, 0x84, 0x1a, 0xbe, 0x23, 0xb1, 0x0d, 0x84,
+	0x78, 0xa3, 0x38, 0x20, 0x4f, 0xe9, 0xe5, 0xf1, 0xc7, 0x51, 0x1a, 0x5f, 0x48, 0x20, 0x74, 0x75,
+	0xbe, 0x06, 0x4b, 0x09, 0x92, 0x36, 0x54, 0xa7, 0xf4, 0x82, 0x4f, 0x36, 0x7b, 0x24, 0xef, 0xc1,
+	0xfa, 0xb9, 0x37, 0x5b, 0x8a, 0x89, 0x6e, 0xe9, 0xe0, 0x9c, 0x9a, 0x9b, 0xe7, 0xec, 0x57, 0x3e,
+	0x31, 0x9c, 0x9f, 0x0d, 0xb0, 0x8e, 0x67, 0xcb, 0x24, 0xa5, 0x31, 0xb2, 0xfe, 0x14, 0xea, 0x9c,
+	0x87, 0x38, 0xe0, 0x4e, 0x4f, 0xde, 0x3d, 0x1e, 0x71, 0xa9, 0xcf, 0x62, 0xd9, 0x3f, 0x91, 0x4e,
+	0x0e, 0xa0, 0xf1, 0x84, 0x8d, 0xb5, 0xfe, 0xd8, 0x45, 0xad, 0x08, 0x69, 0xc5, 0x45, 0x81, 0x73,
+	0x69, 0xc0, 0x66, 0x66, 0xfd, 0x6b, 0xff, 0x77, 0xc1, 0x72, 0xbd, 0x51, 0x7a, 0x18, 0x04, 0x31,
+	0x4d, 0x12, 0x24, 0xd9, 0x70, 0x55, 0x17, 0x79, 0x07, 0x5a, 0x5c, 0x73, 0x22, 0xa9, 0x8a, 0x49,
+	0x25, 0x2f, 0x79, 0x1b, 0x9a, 0x9f, 0xb3, 0x24, 0x99, 0x2c, 0x44, 0x9a, 0x89, 0x69, 0xba, 0x93,
+	0x1c, 0x40, 0x9d, 0xdf, 0xe0, 0xc4, 0x5e, 0x47, 0x46, 0x1d, 0x75, 0xe2, 0x4f, 0xd8, 0x58, 0xec,
+	0x8e, 0x62, 0xe8, 0xb2, 0xc2, 0xf9, 0xad, 0x02, 0x75, 0xa4, 0x94, 0xcd, 0xfd, 0x40, 0x4a, 0x27,
+	0x6f, 0x6b, 0xb7, 0x18, 0x8d, 0xc8, 0xe9, 0xe5, 0x09, 0xfa, 0xe0, 0x45, 0x6f, 0x85, 0x6a, 0x2a,
+	0x37, 0x57, 0xdf, 0x28, 0x9b, 0x21, 0x58, 0x0a, 0xb4, 0x2a, 0x1b, 0x33, 0x97, 0x4d, 0x4f, 0x97,
+	0x8d, 0x5d, 0x22, 0x29, 0x75, 0xad, 0x28, 0xa7, 0xf3, 0xcd, 0x7f, 0x69, 0xf1, 0x9e, 0x0e, 0xba,
+	0x5d, 0x3a, 0xf2, 0x0a, 0x31, 0xfe, 0x6e, 0x40, 0xf3, 0x8b, 0xc3, 0x2f, 0x31, 0x29, 0xef, 0xda,
+	0x2a, 0x11, 0x3c, 0xd2, 0x14, 0x7b, 0x5d, 0xe9, 0x4a, 0x90, 0xf7, 0x41, 0x53, 0xf8, 0x03, 0x79,
+	0x17, 0x51, 0x1a, 0xda, 0xf6, 0xe6, 0x01, 0xb1, 0xbd, 0xc5, 0x9d, 0xfd, 0xb0, 0x98, 0x23, 0x5f,
+	0x5a, 0xe4, 0x7a, 0xff, 0xc5, 0xf8, 0x85, 0xed, 0xfc, 0x6a, 0x42, 0xdd, 0x1d, 0x7e, 0x75, 0x33,
+	0x91, 0x6d, 0xa8, 0x9d, 0xd2, 0x1f, 0x53, 0xf9, 0x91, 0xe2, 0x96, 0xdc, 0x6c, 0x55, 0x65, 0xb3,
+	0x7d, 0x0b, 0xed, 0xd2, 0xfe, 0xca, 0x24, 0x9b, 0x49, 0x61, 0x4f, 0xf9, 0xf8, 0xf0, 0xb7, 0x95,
+	0x57, 0x9d, 0x26, 0x8a, 0x6b, 0x38, 0xe4, 0x33, 0xf5, 0xe2, 0xe6, 0x32, 0x7f, 0x6b, 0x05, 0xa8,
+	0xcc, 0x41, 0x34, 0xe5, 0xee, 0xaa, 0x2d, 0xad, 0xbd, 0x42, 0x4b, 0x37, 0xfe, 0x6f, 0x4b, 0xcb,
+	0xa3, 0xaf, 0xbf, 0xdc, 0xe8, 0x3b, 0xdf, 0xc1, 0xd6, 0xca, 0xce, 0xbc, 0x94, 0x78, 0xd5, 0x8f,
+	0x90, 0x7a, 0x1f, 0x0e, 0xa0, 0xa5, 0x77, 0x68, 0x05, 0xea, 0x1d, 0x15, 0xd5, 0x54, 0xaa, 0xef,
+	0x3e, 0x87, 0x76, 0xf9, 0x27, 0x80, 0xb4, 0x00, 0x0e, 0x83, 0x80, 0xaf, 0x93, 0xf6, 0x1a, 0xb9,
+	0x0d, 0x4d, 0x97, 0xce, 0xd9, 0x39, 0x15, 0x2e, 0x83, 0xb4, 0x61, 0x73, 0x98, 0x7a, 0x71, 0x2a,
+	0x3c, 0x15, 0x72, 0x0b, 0xaf, 0xe5, 0x42, 0x38, 0xaa, 0x77, 0xef, 0x69, 0xdf, 0xeb, 0x0c, 0x34,
+	0x3b, 0x66, 0xee, 0x69, 0xaf, 0x91, 0x26, 0x34, 0x06, 0x91, 0x30, 0x8d, 0xa3, 0xf6, 0xe5, 0xdf,
+	0xdd, 0xb5, 0x17, 0x57, 0x5d, 0xe3, 0xf2, 0xaa, 0x6b, 0xfc, 0x75, 0xd5, 0x35, 0xce, 0x6a, 0xf8,
+	0xc7, 0xf5, 0xc1, 0x3f, 0x01, 0x00, 0x00, 0xff, 0xff, 0x4b, 0x64, 0x65, 0xe9, 0xde, 0x09, 0x00,
 	0x00,
 }
 
@@ -989,6 +1050,40 @@ func (m *ConfigChange) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *ShutdownStore) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *ShutdownStore) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *ShutdownStore) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.StoreID) > 0 {
+		i -= len(m.StoreID)
+		copy(dAtA[i:], m.StoreID)
+		i = encodeVarintHakeeper(dAtA, i, uint64(len(m.StoreID)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *ScheduleCommand) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1013,21 +1108,35 @@ func (m *ScheduleCommand) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.ShutdownStore != nil {
+		{
+			size, err := m.ShutdownStore.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintHakeeper(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x22
+	}
 	if m.ServiceType != 0 {
 		i = encodeVarintHakeeper(dAtA, i, uint64(m.ServiceType))
 		i--
 		dAtA[i] = 0x18
 	}
-	{
-		size, err := m.ConfigChange.MarshalToSizedBuffer(dAtA[:i])
-		if err != nil {
-			return 0, err
+	if m.ConfigChange != nil {
+		{
+			size, err := m.ConfigChange.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintHakeeper(dAtA, i, uint64(size))
 		}
-		i -= size
-		i = encodeVarintHakeeper(dAtA, i, uint64(size))
+		i--
+		dAtA[i] = 0x12
 	}
-	i--
-	dAtA[i] = 0x12
 	if len(m.UUID) > 0 {
 		i -= len(m.UUID)
 		copy(dAtA[i:], m.UUID)
@@ -1604,6 +1713,22 @@ func (m *ConfigChange) Size() (n int) {
 	return n
 }
 
+func (m *ShutdownStore) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.StoreID)
+	if l > 0 {
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *ScheduleCommand) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1614,11 +1739,17 @@ func (m *ScheduleCommand) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovHakeeper(uint64(l))
 	}
-	l = m.ConfigChange.Size()
-	n += 1 + l + sovHakeeper(uint64(l))
+	if m.ConfigChange != nil {
+		l = m.ConfigChange.Size()
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
 	if m.ServiceType != 0 {
 		n += 1 + sovHakeeper(uint64(m.ServiceType))
 	}
+	if m.ShutdownStore != nil {
+		l = m.ShutdownStore.Size()
+		n += 1 + l + sovHakeeper(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2089,6 +2220,89 @@ func (m *ConfigChange) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *ShutdownStore) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowHakeeper
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: ShutdownStore: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: ShutdownStore: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field StoreID", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.StoreID = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipHakeeper(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
@@ -2179,6 +2393,9 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
+			if m.ConfigChange == nil {
+				m.ConfigChange = &ConfigChange{}
+			}
 			if err := m.ConfigChange.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
 				return err
 			}
@@ -2202,6 +2419,42 @@ func (m *ScheduleCommand) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
+		case 4:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ShutdownStore", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowHakeeper
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthHakeeper
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.ShutdownStore == nil {
+				m.ShutdownStore = &ShutdownStore{}
+			}
+			if err := m.ShutdownStore.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipHakeeper(dAtA[iNdEx:])
diff --git a/pkg/sql/plan2/function/operator/eq.go b/pkg/sql/plan2/function/operator/eq.go
index b40a853dc..c35ef3a56 100644
--- a/pkg/sql/plan2/function/operator/eq.go
+++ b/pkg/sql/plan2/function/operator/eq.go
@@ -80,6 +80,22 @@ func ScalarEqNotScalar[T NormalType](sv, nsv *vector.Vector, col1, col2 []T, pro
 	return vec, nil
 }
 
+func ScalarDecimal128EqNotScalar(sv, nsv *vector.Vector, str types.Decimal128, col []types.Decimal128, proc *process.Process) (*vector.Vector, error) {
+	var i int64
+	length := int64(vector.Length(nsv))
+	vec, err := allocateBoolVector(length, proc)
+	if err != nil {
+		return nil, err
+	}
+	vcols := vec.Col.([]bool)
+	for i = 0; i < length; i++ {
+		vcols[i] = types.CompareDecimal128Decimal128(str, col[i], sv.Typ.Scale, nsv.Typ.Scale) == 0
+	}
+	nulls.Or(nsv.Nsp, nil, vec.Nsp)
+	FillNullPos(vec)
+	return vec, nil
+}
+
 func ScalarStringEqNotScalar(sv, nsv *vector.Vector, str []byte, col *types.Bytes, proc *process.Process) (*vector.Vector, error) {
 	var i int64
 	length := int64(vector.Length(nsv))
@@ -136,6 +152,42 @@ func EqGeneral[T NormalType](vs []*vector.Vector, proc *process.Process) (*vecto
 	return vec, nil
 }
 
+func EqDecimal128(vs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	v1, v2 := vs[0], vs[1]
+	col1, col2 := vector.MustTCols[types.Decimal128](v1), vector.MustTCols[types.Decimal128](v2)
+
+	if v1.IsScalarNull() || v2.IsScalarNull() {
+		return HandleWithNullCol(vs, proc)
+	}
+
+	c1, c2 := v1.IsScalar(), v2.IsScalar()
+	switch {
+	case c1 && c2:
+		vec := proc.AllocScalarVector(retType)
+		vec.Col = make([]bool, 1)
+		vec.Col.([]bool)[0] = types.CompareDecimal128Decimal128(col1[0], col2[0], v1.Typ.Scale, v2.Typ.Scale) == 0
+		return vec, nil
+	case c1 && !c2:
+		return ScalarDecimal128EqNotScalar(v1, v2, col1[0], col2, proc)
+	case !c1 && c2:
+		return ScalarDecimal128EqNotScalar(v2, v1, col2[0], col1, proc)
+	}
+	// case !c1 && !c2
+	length := int64(vector.Length(v1))
+	vec, err := allocateBoolVector(length, proc)
+	if err != nil {
+		return nil, err
+	}
+	vcols := vec.Col.([]bool)
+	for i := range vcols {
+		j := int64(i)
+		vcols[i] = types.CompareDecimal128Decimal128(col1[j], col2[j], v1.Typ.Scale, v2.Typ.Scale) == 0
+	}
+	nulls.Or(v1.Nsp, v2.Nsp, vec.Nsp)
+	FillNullPos(vec)
+	return vec, nil
+}
+
 func EqString(vs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
 	v1, v2 := vs[0], vs[1]
 	col1, col2 := vector.MustBytesCols(v1), vector.MustBytesCols(v2)
diff --git a/pkg/sql/plan2/function/operators.go b/pkg/sql/plan2/function/operators.go
index f41eb3c34..084e57e34 100644
--- a/pkg/sql/plan2/function/operators.go
+++ b/pkg/sql/plan2/function/operators.go
@@ -15,12 +15,13 @@
 package function
 
 import (
+	"math"
+
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan2/function/operator"
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
-	"math"
 )
 
 func initOperators() {
@@ -557,7 +558,7 @@ var operators = map[int]Functions{
 					types.T_decimal128,
 				},
 				ReturnTyp: types.T_bool,
-				Fn:        operator.EqGeneral[types.Decimal128],
+				Fn:        operator.EqDecimal128,
 			},
 			{
 				Index:  12,
diff --git a/proto/gen.sh b/proto/gen.sh
index cf5fb5b24..abbc2baff 100755
--- a/proto/gen.sh
+++ b/proto/gen.sh
@@ -13,4 +13,5 @@ do
 	dir=$(basename $file .proto)
 	mkdir -p $PB_DIR/$dir
 	protoc -I=.:$PROTOC_DIR:$VENDOR_DIR --gogofast_out=paths=source_relative:./pkg/pb/$dir  $file
+    goimports -w $PB_DIR/$dir/*pb.go
 done
diff --git a/proto/hakeeper.proto b/proto/hakeeper.proto
index 05fcf775c..98f124752 100644
--- a/proto/hakeeper.proto
+++ b/proto/hakeeper.proto
@@ -25,106 +25,117 @@ option (gogoproto.goproto_enum_prefix_all) = false;
 
 // Replica of the shard
 message Replica {
-  // UUID which store the Replica is located in
-  string UUID      = 1;
+    // UUID which store the Replica is located in
+    string UUID = 1;
 
-  uint64 ShardID   = 2;
-  uint64 ReplicaID = 3;
-  uint64 Epoch     = 4;
+    uint64 ShardID   = 2;
+    uint64 ReplicaID = 3;
+    uint64 Epoch     = 4;
 }
 
 // ConfigChangeType indicates config change command type.
 enum ConfigChangeType {
-  AddReplica    = 0;
-  RemoveReplica = 1;
-  StartReplica  = 2;
-  StopReplica   = 3;
+    AddReplica    = 0;
+    RemoveReplica = 1;
+    StartReplica  = 2;
+    StopReplica   = 3;
 }
 
 // ConfigChange is the detail of a config change.
 message ConfigChange {
-  Replica Replica             = 1 [(gogoproto.nullable) = false];
-  ConfigChangeType ChangeType = 2;
+    Replica Replica             = 1 [(gogoproto.nullable) = false];
+    ConfigChangeType ChangeType = 2;
+}
+
+// ShutdownStore would stop store.
+message ShutdownStore {
+    string StoreID = 1;
 }
 
 // ServiceType specifies type of service
 enum ServiceType {
-  LogService = 0;
-  DnService  = 1;
+    LogService = 0;
+    DnService  = 1;
 }
 
 // ScheduleCommand contains a shard schedule command.
 message ScheduleCommand {
-  // UUID which store the ScheduleCommand is sent to
-  string UUID               = 1;
+    // UUID which store the ScheduleCommand is sent to
+    string UUID = 1;
 
-  ConfigChange ConfigChange = 2 [(gogoproto.nullable) = false];
-  ServiceType ServiceType   = 3;
+    ConfigChange ConfigChange   = 2;
+    ServiceType ServiceType     = 3;
+    ShutdownStore ShutdownStore = 4;
 }
 
 message CommandBatch {
-  uint64 Term = 1;
-  repeated ScheduleCommand Commands = 2 [(gogoproto.nullable) = false]; 
+    uint64 Term                       = 1;
+    repeated ScheduleCommand Commands = 2 [(gogoproto.nullable) = false];
 }
 
 // DNStoreInfo contins information on a list of shards.
 message DNStoreInfo {
-  uint64 Tick = 1;
-  repeated logservice.DNShardInfo Shards = 2 [(gogoproto.nullable) = false];
+    uint64 Tick                            = 1;
+    repeated logservice.DNShardInfo Shards = 2 [(gogoproto.nullable) = false];
 }
 
 // DNState contains all DN details known to the HAKeeper.
 message DNState {
-  // Stores is keyed by DN store UUID, it contains details found on each DN
-  // store. Each DNStoreInfo reflects what was last reported by each DN store.
-  map<string, DNStoreInfo> Stores = 1 [(gogoproto.nullable) = false];
+    // Stores is keyed by DN store UUID, it contains details found on each DN
+    // store. Each DNStoreInfo reflects what was last reported by each DN store.
+    map<string, DNStoreInfo> Stores = 1 [(gogoproto.nullable) = false];
 }
 
 // ClusterInfo provides a global view of all shards in the cluster. It
 // describes the logical sharding of the system, rather than physical
 // distribution of all replicas that belong to those shards.
 message ClusterInfo {
-  repeated metadata.DNShardRecord DNShards = 1 [(gogoproto.nullable) = false];
-  repeated metadata.LogShardRecord LogShards = 2 [(gogoproto.nullable) = false];
+    repeated metadata.DNShardRecord DNShards = 1 [(gogoproto.nullable) = false];
+    repeated metadata.LogShardRecord LogShards = 2
+        [(gogoproto.nullable) = false];
 }
 
 // LogStoreInfo contains information of all replicas found on a Log store.
 message LogStoreInfo {
-  uint64 Tick = 1;
-  string RaftAddress = 2;
-  string ServiceAddress = 3;
-  string GossipAddress = 4;
-  repeated logservice.LogReplicaInfo Replicas = 5 [(gogoproto.nullable) = false];
+    uint64 Tick                                 = 1;
+    string RaftAddress                          = 2;
+    string ServiceAddress                       = 3;
+    string GossipAddress                        = 4;
+    repeated logservice.LogReplicaInfo Replicas = 5
+        [(gogoproto.nullable) = false];
 }
 
 message LogState {
-  // Shards is keyed by ShardID, it contains details aggregated from all Log
-  // stores. Each pb.LogShardInfo here contains data aggregated from
-  // different replicas and thus reflect a more accurate description on each
-  // shard.
-  map<uint64, logservice.LogShardInfo> Shards = 1 [(gogoproto.nullable) = false];
-  // Stores is keyed by log store UUID, it contains details found on each store.
-  // Each LogStoreInfo here reflects what was last reported by each Log store.
-  map<string, LogStoreInfo> Stores = 2 [(gogoproto.nullable) = false];
+    // Shards is keyed by ShardID, it contains details aggregated from all Log
+    // stores. Each pb.LogShardInfo here contains data aggregated from
+    // different replicas and thus reflect a more accurate description on each
+    // shard.
+    map<uint64, logservice.LogShardInfo> Shards = 1
+        [(gogoproto.nullable) = false];
+    // Stores is keyed by log store UUID, it contains details found on each
+    // store. Each LogStoreInfo here reflects what was last reported by each Log
+    // store.
+    map<string, LogStoreInfo> Stores = 2 [(gogoproto.nullable) = false];
 }
 
 // HAKeeperState contains all HAKeeper state required for making schedule
 // commands.
 message HAKeeperState {
-  uint64 Tick = 1;
-  ClusterInfo ClusterInfo = 2 [(gogoproto.nullable) = false];
-  DNState DNState = 3 [(gogoproto.nullable) = false];
-  LogState LogState = 4 [(gogoproto.nullable) = false];
+    uint64 Tick             = 1;
+    ClusterInfo ClusterInfo = 2 [(gogoproto.nullable) = false];
+    DNState DNState         = 3 [(gogoproto.nullable) = false];
+    LogState LogState       = 4 [(gogoproto.nullable) = false];
 }
 
-// RSMState contains state maintained by HAKeeper's RSM.  
+// RSMState contains state maintained by HAKeeper's RSM.
 message RSMState {
-  uint64 Tick = 1;
-  uint64 NextID = 2;
-  uint64 Term = 3;
-  map<string, CommandBatch> ScheduleCommands = 4 [(gogoproto.nullable) = false];
-  map<string, uint64> LogShards = 5;
-  DNState DNState = 6 [(gogoproto.nullable) = false];
-  LogState LogState = 7 [(gogoproto.nullable) = false];
-  ClusterInfo ClusterInfo = 8 [(gogoproto.nullable) = false];
+    uint64 Tick                                = 1;
+    uint64 NextID                              = 2;
+    uint64 Term                                = 3;
+    map<string, CommandBatch> ScheduleCommands = 4
+        [(gogoproto.nullable) = false];
+    map<string, uint64> LogShards = 5;
+    DNState DNState               = 6 [(gogoproto.nullable) = false];
+    LogState LogState             = 7 [(gogoproto.nullable) = false];
+    ClusterInfo ClusterInfo       = 8 [(gogoproto.nullable) = false];
 }


Test patch:
diff --git a/pkg/hakeeper/checkers/coordinator_test.go b/pkg/hakeeper/checkers/coordinator_test.go
index 2155a6222..be6c65eca 100644
--- a/pkg/hakeeper/checkers/coordinator_test.go
+++ b/pkg/hakeeper/checkers/coordinator_test.go
@@ -15,13 +15,14 @@ package checkers
 
 import (
 	"fmt"
+	"testing"
+
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
 	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"github.com/stretchr/testify/assert"
-	"testing"
 )
 
 func TestFixExpiredStore(t *testing.T) {
@@ -142,7 +143,7 @@ func TestFixExpiredStore(t *testing.T) {
 			currentTick: 15 * hakeeper.TickPerSecond * 60,
 			expected: []hapb.ScheduleCommand{{
 				UUID: "b",
-				ConfigChange: hapb.ConfigChange{
+				ConfigChange: &hapb.ConfigChange{
 					Replica: hapb.Replica{
 						UUID:      "a",
 						ShardID:   1,
@@ -203,7 +204,7 @@ func TestFixExpiredStore(t *testing.T) {
 			currentTick: 15 * hakeeper.TickPerSecond * 60,
 			expected: []hapb.ScheduleCommand{{
 				UUID: "b",
-				ConfigChange: hapb.ConfigChange{
+				ConfigChange: &hapb.ConfigChange{
 					Replica: hapb.Replica{
 						UUID:      "a",
 						ShardID:   1,
@@ -265,7 +266,7 @@ func TestFixExpiredStore(t *testing.T) {
 			expected: []hapb.ScheduleCommand{
 				{
 					UUID: "a",
-					ConfigChange: hapb.ConfigChange{
+					ConfigChange: &hapb.ConfigChange{
 						Replica: hapb.Replica{
 							UUID:      "a",
 							ShardID:   1,
@@ -364,7 +365,7 @@ func TestFixZombie(t *testing.T) {
 			expected: []hapb.ScheduleCommand{
 				{
 					UUID: "c",
-					ConfigChange: hapb.ConfigChange{
+					ConfigChange: &hapb.ConfigChange{
 						Replica: hapb.Replica{
 							UUID:    "c",
 							ShardID: 1,
diff --git a/pkg/hakeeper/checkers/syshealth/check_test.go b/pkg/hakeeper/checkers/syshealth/check_test.go
new file mode 100644
index 000000000..633389285
--- /dev/null
+++ b/pkg/hakeeper/checkers/syshealth/check_test.go
@@ -0,0 +1,329 @@
+// 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 syshealth
+
+import (
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
+	hapb "github.com/matrixorigin/matrixone/pkg/pb/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/pb/logservice"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestShutdownStores(t *testing.T) {
+	stores := map[util.StoreID]struct{}{
+		util.StoreID("11"): {},
+		util.StoreID("12"): {},
+		util.StoreID("13"): {},
+	}
+
+	// operator for log service
+	{
+		serviceType := hapb.LogService
+		ops := shutdownStores(serviceType, stores)
+		require.Equal(t, len(stores), len(ops))
+
+		for i := 0; i < len(ops); i++ {
+			op := ops[i]
+			steps := op.OpSteps()
+			require.Equal(t, 1, len(steps))
+
+			_, ok := steps[0].(operator.StopLogStore)
+			require.True(t, ok)
+		}
+	}
+
+	// operator for dn service
+	{
+		serviceType := hapb.DnService
+		ops := shutdownStores(serviceType, stores)
+		require.Equal(t, len(stores), len(ops))
+
+		for i := 0; i < len(ops); i++ {
+			op := ops[i]
+			steps := op.OpSteps()
+			require.Equal(t, 1, len(steps))
+
+			_, ok := steps[0].(operator.StopDnStore)
+			require.True(t, ok)
+		}
+	}
+}
+
+func TestParseLogStores(t *testing.T) {
+	expiredTick := uint64(10)
+	// construct current tick in order to make hearbeat tick expired
+	currTick := hakeeper.ExpiredTick(expiredTick, hakeeper.LogStoreTimeout) + 1
+
+	logState := hapb.LogState{
+		Stores: map[string]hapb.LogStoreInfo{
+			"expired1": mockLogStoreInfo(
+				expiredTick,
+				mockLogReplicaInfo(10, 100),
+				mockLogReplicaInfo(11, 101),
+			),
+			"working1": mockLogStoreInfo(
+				currTick,
+				mockLogReplicaInfo(10, 102),
+				mockLogReplicaInfo(11, 103),
+			),
+			"working2": mockLogStoreInfo(
+				currTick,
+				mockLogReplicaInfo(10, 104),
+				mockLogReplicaInfo(11, 105),
+			),
+		},
+	}
+
+	logStores := parseLogStores(logState, currTick)
+	require.Equal(t, len(logState.Stores), logStores.length())
+	require.Equal(t, hapb.LogService, logStores.serviceType)
+	require.Equal(t, 1, len(logStores.expired))
+	require.Equal(t, 1, len(logStores.shutdownExpiredStores()))
+	require.Equal(t, 2, len(logStores.working))
+	require.Equal(t, 2, len(logStores.shutdownWorkingStores()))
+}
+
+func TestParseDnStores(t *testing.T) {
+	expiredTick := uint64(10)
+	// construct current tick in order to make hearbeat tick expired
+	currTick := hakeeper.ExpiredTick(expiredTick, hakeeper.LogStoreTimeout) + 1
+
+	dnState := hapb.DNState{
+		Stores: map[string]hapb.DNStoreInfo{
+			"expired1": {
+				Tick: expiredTick,
+				Shards: []logservice.DNShardInfo{
+					mockDnShardInfo(10, 100),
+				},
+			},
+			"working1": {
+				Tick: currTick,
+				Shards: []logservice.DNShardInfo{
+					mockDnShardInfo(11, 101),
+				},
+			},
+			"working2": {
+				Tick: currTick,
+				Shards: []logservice.DNShardInfo{
+					mockDnShardInfo(12, 102),
+					mockDnShardInfo(13, 103),
+				},
+			},
+		},
+	}
+
+	dnStores := parseDnStores(dnState, currTick)
+	require.Equal(t, len(dnState.Stores), dnStores.length())
+	require.Equal(t, hapb.DnService, dnStores.serviceType)
+	require.Equal(t, 1, len(dnStores.expired))
+	require.Equal(t, 1, len(dnStores.shutdownExpiredStores()))
+	require.Equal(t, 2, len(dnStores.working))
+	require.Equal(t, 2, len(dnStores.shutdownWorkingStores()))
+}
+
+func TestLogShard(t *testing.T) {
+	// odd shard size
+	{
+		logShard := newLogShard(10, defaultLogShardSize)
+		require.True(t, logShard.healthy())
+
+		logShard.registerExpiredReplica(100)
+		require.True(t, logShard.healthy())
+
+		logShard.registerExpiredReplica(101)
+		require.False(t, logShard.healthy())
+	}
+
+	// even shard size
+	{
+		logShard := newLogShard(10, 2)
+		require.True(t, logShard.healthy())
+
+		logShard.registerExpiredReplica(100)
+		require.False(t, logShard.healthy())
+	}
+}
+
+func TestLogShardMap(t *testing.T) {
+	expiredStores := map[util.StoreID]struct{}{
+		"expired1": {},
+		"expired2": {},
+		"expired3": {},
+	}
+
+	tick := uint64(10)
+
+	logState := hapb.LogState{
+		Stores: map[string]hapb.LogStoreInfo{
+			"expired1": mockLogStoreInfo(
+				tick,
+				mockLogReplicaInfo(10, 100),
+				mockLogReplicaInfo(11, 101),
+			),
+			"expired2": mockLogStoreInfo(
+				tick,
+				mockLogReplicaInfo(10, 102),
+				mockLogReplicaInfo(12, 103),
+			),
+			"expired3": mockLogStoreInfo(
+				tick,
+				mockLogReplicaInfo(13, 104),
+				mockLogReplicaInfo(14, 105),
+			),
+		},
+	}
+
+	logShardMap := logShardsWithExpired(expiredStores, logState, hapb.ClusterInfo{})
+	require.Equal(t, 5, len(logShardMap))
+	require.Equal(t, 2, len(logShardMap[10].expiredReplicas))
+	require.Equal(t, 1, len(logShardMap[11].expiredReplicas))
+	require.Equal(t, 1, len(logShardMap[12].expiredReplicas))
+	require.Equal(t, 1, len(logShardMap[13].expiredReplicas))
+	require.Equal(t, 1, len(logShardMap[14].expiredReplicas))
+}
+
+func TestCheck(t *testing.T) {
+	expiredTick := uint64(10)
+	// construct current tick in order to make hearbeat tick expired
+	currTick := hakeeper.ExpiredTick(expiredTick, hakeeper.LogStoreTimeout) + 1
+
+	// system healthy
+	{
+		logState := hapb.LogState{
+			Stores: map[string]hapb.LogStoreInfo{
+				"expired1": mockLogStoreInfo(
+					expiredTick,
+					mockLogReplicaInfo(10, 100),
+					mockLogReplicaInfo(11, 101),
+				),
+				"working1": mockLogStoreInfo(
+					currTick,
+					mockLogReplicaInfo(10, 102),
+					mockLogReplicaInfo(11, 103),
+				),
+				"working2": mockLogStoreInfo(
+					currTick,
+					mockLogReplicaInfo(10, 104),
+					mockLogReplicaInfo(11, 105),
+				),
+			},
+		}
+
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
+				"expired11": {
+					Tick: expiredTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(10, 100),
+					},
+				},
+				"working11": {
+					Tick: currTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(11, 101),
+					},
+				},
+				"working12": {
+					Tick: currTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(12, 102),
+						mockDnShardInfo(13, 103),
+					},
+				},
+			},
+		}
+
+		ops, healthy := Check(hapb.ClusterInfo{}, dnState, logState, currTick)
+		require.True(t, healthy)
+		require.Equal(t, 0, len(ops))
+	}
+
+	// system unhealthy
+	{
+		logState := hapb.LogState{
+			Stores: map[string]hapb.LogStoreInfo{
+				"expired1": mockLogStoreInfo(
+					expiredTick,
+					mockLogReplicaInfo(10, 100),
+					mockLogReplicaInfo(11, 101),
+				),
+				"expired2": mockLogStoreInfo(
+					expiredTick,
+					mockLogReplicaInfo(10, 102),
+					mockLogReplicaInfo(11, 103),
+				),
+				"working2": mockLogStoreInfo(
+					currTick,
+					mockLogReplicaInfo(10, 104),
+					mockLogReplicaInfo(11, 105),
+				),
+			},
+		}
+
+		dnState := hapb.DNState{
+			Stores: map[string]hapb.DNStoreInfo{
+				"expired11": {
+					Tick: expiredTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(10, 100),
+					},
+				},
+				"working11": {
+					Tick: currTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(11, 101),
+					},
+				},
+				"working12": {
+					Tick: currTick,
+					Shards: []logservice.DNShardInfo{
+						mockDnShardInfo(12, 102),
+						mockDnShardInfo(13, 103),
+					},
+				},
+			},
+		}
+
+		ops, healthy := Check(hapb.ClusterInfo{}, dnState, logState, currTick)
+		require.False(t, healthy)
+		require.Equal(t, 6, len(ops))
+	}
+}
+
+func mockLogReplicaInfo(shardID, replicaID uint64) logservice.LogReplicaInfo {
+	info := logservice.LogReplicaInfo{
+		ReplicaID: replicaID,
+	}
+	info.ShardID = shardID
+	return info
+}
+
+func mockLogStoreInfo(tick uint64, replicas ...logservice.LogReplicaInfo) hapb.LogStoreInfo {
+	return hapb.LogStoreInfo{
+		Tick:     tick,
+		Replicas: replicas,
+	}
+}
+
+func mockDnShardInfo(shardID, replicaID uint64) logservice.DNShardInfo {
+	return logservice.DNShardInfo{
+		ShardID:   shardID,
+		ReplicaID: replicaID,
+	}
+}
diff --git a/pkg/hakeeper/rsm_test.go b/pkg/hakeeper/rsm_test.go
index a0996b182..b46fc8eac 100644
--- a/pkg/hakeeper/rsm_test.go
+++ b/pkg/hakeeper/rsm_test.go
@@ -234,7 +234,7 @@ func TestUpdateScheduleCommandsCmd(t *testing.T) {
 	tsm1 := NewStateMachine(0, 1).(*stateMachine)
 	sc1 := hapb.ScheduleCommand{
 		UUID: "uuid1",
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			Replica: hapb.Replica{
 				ShardID: 1,
 			},
@@ -242,7 +242,7 @@ func TestUpdateScheduleCommandsCmd(t *testing.T) {
 	}
 	sc2 := hapb.ScheduleCommand{
 		UUID: "uuid2",
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			Replica: hapb.Replica{
 				ShardID: 2,
 			},
@@ -250,7 +250,7 @@ func TestUpdateScheduleCommandsCmd(t *testing.T) {
 	}
 	sc3 := hapb.ScheduleCommand{
 		UUID: "uuid1",
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			Replica: hapb.Replica{
 				ShardID: 3,
 			},
@@ -258,7 +258,7 @@ func TestUpdateScheduleCommandsCmd(t *testing.T) {
 	}
 	sc4 := hapb.ScheduleCommand{
 		UUID: "uuid3",
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			Replica: hapb.Replica{
 				ShardID: 4,
 			},
diff --git a/pkg/logservice/commands_test.go b/pkg/logservice/commands_test.go
index 900c86101..c9dc6317d 100644
--- a/pkg/logservice/commands_test.go
+++ b/pkg/logservice/commands_test.go
@@ -27,7 +27,7 @@ import (
 func TestHandleStartReplica(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		cmd := hapb.ScheduleCommand{
-			ConfigChange: hapb.ConfigChange{
+			ConfigChange: &hapb.ConfigChange{
 				ChangeType: hapb.StartReplica,
 				Replica: hapb.Replica{
 					ShardID:   1,
@@ -44,7 +44,7 @@ func TestHandleStartReplica(t *testing.T) {
 func TestHandleStopReplica(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		cmd := hapb.ScheduleCommand{
-			ConfigChange: hapb.ConfigChange{
+			ConfigChange: &hapb.ConfigChange{
 				ChangeType: hapb.StartReplica,
 				Replica: hapb.Replica{
 					ShardID:   1,
@@ -56,7 +56,7 @@ func TestHandleStopReplica(t *testing.T) {
 		mustHaveReplica(t, s.store, 1, 1)
 
 		cmd = hapb.ScheduleCommand{
-			ConfigChange: hapb.ConfigChange{
+			ConfigChange: &hapb.ConfigChange{
 				ChangeType: hapb.StopReplica,
 				Replica: hapb.Replica{
 					ShardID:   1,
@@ -82,7 +82,7 @@ func TestHandleAddReplica(t *testing.T) {
 		store: store1,
 	}
 	cmd := hapb.ScheduleCommand{
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			ChangeType: hapb.AddReplica,
 			Replica: hapb.Replica{
 				UUID:      uuid.New().String(),
@@ -110,7 +110,7 @@ func TestHandleRemoveReplica(t *testing.T) {
 		store: store1,
 	}
 	cmd := hapb.ScheduleCommand{
-		ConfigChange: hapb.ConfigChange{
+		ConfigChange: &hapb.ConfigChange{
 			ChangeType: hapb.RemoveReplica,
 			Replica: hapb.Replica{
 				ShardID:   1,