Repository: matrixorigin/matrixone
Pull request #4327 (2022-08-02 09:02:20)

logservice: stop hakeeper ticker on KillZombie command
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

When HAKeeper replica is stopped by a KillZombie command, HAKeeper ticker managed by the Log Store should be stopped as well. This avoids flooding the log with a ton of ticker failed error messages.  

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 2507ace80..0673fe408 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -111,6 +111,7 @@ type store struct {
 	checker           hakeeper.Checker
 	alloc             hakeeper.IDAllocator
 	stopper           *stopper.Stopper
+	tickerStopper     *stopper.Stopper
 
 	bootstrapCheckCycles uint64
 	bootstrapMgr         *bootstrap.Manager
@@ -132,11 +133,12 @@ func newLogStore(cfg Config) (*store, error) {
 	plog.Infof("HAKeeper LogStoreTimeout: %s, DnStoreTimeout: %s",
 		hakeeperConfig.LogStoreTimeout, hakeeperConfig.DnStoreTimeout)
 	ls := &store{
-		cfg:     cfg,
-		nh:      nh,
-		checker: checkers.NewCoordinator(hakeeperConfig),
-		alloc:   newIDAllocator(),
-		stopper: stopper.NewStopper("log-store"),
+		cfg:           cfg,
+		nh:            nh,
+		checker:       checkers.NewCoordinator(hakeeperConfig),
+		alloc:         newIDAllocator(),
+		stopper:       stopper.NewStopper("log-store"),
+		tickerStopper: stopper.NewStopper("hakeeper-ticker"),
 	}
 	ls.mu.truncateCh = make(chan struct{})
 	ls.mu.pendingTruncate = make(map[uint64]struct{})
@@ -151,6 +153,7 @@ func newLogStore(cfg Config) (*store, error) {
 }
 
 func (l *store) close() error {
+	l.tickerStopper.Stop()
 	l.stopper.Stop()
 	if l.nh != nil {
 		l.nh.Close()
@@ -192,7 +195,7 @@ func (l *store) startHAKeeperReplica(replicaID uint64,
 	l.addMetadata(hakeeper.DefaultHAKeeperShardID, replicaID)
 	atomic.StoreUint64(&l.haKeeperReplicaID, replicaID)
 	if !l.cfg.DisableWorkers {
-		if err := l.stopper.RunNamedTask("hakeeper-ticker", func(ctx context.Context) {
+		if err := l.tickerStopper.RunNamedTask("hakeeper-ticker", func(ctx context.Context) {
 			plog.Infof("HAKeeper ticker started")
 			l.ticker(ctx)
 		}); err != nil {
@@ -216,8 +219,10 @@ func (l *store) startReplica(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
+	if shardID == hakeeper.DefaultHAKeeperShardID {
+		plog.Infof("going to stop HAKeeper's ticker")
+		l.tickerStopper.Stop()
+	}
 	return l.nh.StopReplica(shardID, replicaID)
 }
 


Test patch:
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 4e5e28761..585ee8be1 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -498,6 +498,18 @@ func TestRemoveReplica(t *testing.T) {
 	}
 }
 
+func TestStopReplicaCanStopHAKeeperTicker(t *testing.T) {
+	fn := func(t *testing.T, store *store) {
+		peers := make(map[uint64]dragonboat.Target)
+		peers[1] = store.id()
+		assert.NoError(t, store.startHAKeeperReplica(1, peers, false))
+		assert.Equal(t, int64(1), store.tickerStopper.GetTaskCount())
+		assert.NoError(t, store.stopReplica(hakeeper.DefaultHAKeeperShardID, 1))
+		assert.Equal(t, int64(0), store.tickerStopper.GetTaskCount())
+	}
+	runStoreTest(t, fn)
+}
+
 func hasShard(s *store, shardID uint64) bool {
 	hb := s.getHeartbeatMessage()
 	for _, info := range hb.Replicas {