Repository: matrixorigin/matrixone
Pull request #4313 (2022-08-01 13:37:12)

logservice: handle kill zombie command
**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:**

Zombie replicas are now handled by the dedicated KillZombie command. 

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go
index 687c907ea..8987b86c5 100644
--- a/pkg/logservice/service_commands.go
+++ b/pkg/logservice/service_commands.go
@@ -32,13 +32,13 @@ func (s *Service) handleCommands(cmds []pb.ScheduleCommand) {
 			case pb.AddReplica:
 				s.handleAddReplica(cmd)
 			case pb.RemoveReplica:
-				// FIXME: when remove replica cmd is received, we need to stop the zombie
-				// replica running on the local store.
 				s.handleRemoveReplica(cmd)
 			case pb.StartReplica:
 				s.handleStartReplica(cmd)
 			case pb.StopReplica:
 				s.handleStopReplica(cmd)
+			case pb.KillZombie:
+				s.handleKillZombie(cmd)
 			default:
 				panic("unknown config change cmd type")
 			}
@@ -94,6 +94,13 @@ func (s *Service) handleStopReplica(cmd pb.ScheduleCommand) {
 	}
 }
 
+func (s *Service) handleKillZombie(cmd pb.ScheduleCommand) {
+	shardID := cmd.ConfigChange.Replica.ShardID
+	replicaID := cmd.ConfigChange.Replica.ReplicaID
+	s.handleStopReplica(cmd)
+	s.store.removeMetadata(shardID, replicaID)
+}
+
 func (s *Service) handleShutdownStore(cmd pb.ScheduleCommand) {
 	panic("not implemented")
 }
diff --git a/pkg/logservice/store_metadata.go b/pkg/logservice/store_metadata.go
index 1c4ef6493..6470527eb 100644
--- a/pkg/logservice/store_metadata.go
+++ b/pkg/logservice/store_metadata.go
@@ -221,6 +221,20 @@ func readMetadataFile(dir string,
 	return nil
 }
 
+func hasMetadataRec(dir string,
+	filename string, shardID uint64, replicaID uint64, fs vfs.FS) (has bool, err error) {
+	var md metadata.LogStore
+	if err := readMetadataFile(dir, filename, &md, fs); err != nil {
+		return false, err
+	}
+	for _, rec := range md.Shards {
+		if rec.ShardID == shardID && rec.ReplicaID == replicaID {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func (l *store) loadMetadata() error {
 	fs := l.cfg.FS
 	dir := l.cfg.DataDir


Test patch:
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index ed573f7b0..a39c8b35b 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -78,6 +78,32 @@ func TestBackgroundTickAndHeartbeat(t *testing.T) {
 	t.Fatalf("failed to tick/heartbeat")
 }
 
+func TestHandleKillZombie(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		has, err := hasMetadataRec(s.store.cfg.DataDir, logMetadataFilename, 1, 1, s.store.cfg.FS)
+		require.NoError(t, err)
+		assert.True(t, has)
+
+		cmd := pb.ScheduleCommand{
+			ConfigChange: &pb.ConfigChange{
+				ChangeType: pb.KillZombie,
+				Replica: pb.Replica{
+					ShardID:   1,
+					ReplicaID: 1,
+				},
+			},
+		}
+		mustHaveReplica(t, s.store, 1, 1)
+		s.handleCommands([]pb.ScheduleCommand{cmd})
+		assert.False(t, hasReplica(s.store, 1, 1))
+
+		has, err = hasMetadataRec(s.store.cfg.DataDir, logMetadataFilename, 1, 1, s.store.cfg.FS)
+		require.NoError(t, err)
+		assert.False(t, has)
+	}
+	runServiceTest(t, false, true, fn)
+}
+
 func TestHandleStartReplica(t *testing.T) {
 	fn := func(t *testing.T, s *Service) {
 		cmd := pb.ScheduleCommand{
@@ -92,6 +118,10 @@ func TestHandleStartReplica(t *testing.T) {
 		}
 		s.handleCommands([]pb.ScheduleCommand{cmd})
 		mustHaveReplica(t, s.store, 1, 1)
+
+		has, err := hasMetadataRec(s.store.cfg.DataDir, logMetadataFilename, 1, 1, s.store.cfg.FS)
+		require.NoError(t, err)
+		assert.True(t, has)
 	}
 	runServiceTest(t, false, false, fn)
 }
@@ -122,6 +152,10 @@ func TestHandleStopReplica(t *testing.T) {
 		}
 		s.handleCommands([]pb.ScheduleCommand{cmd})
 		assert.False(t, hasReplica(s.store, 1, 1))
+
+		has, err := hasMetadataRec(s.store.cfg.DataDir, logMetadataFilename, 1, 1, s.store.cfg.FS)
+		require.NoError(t, err)
+		assert.True(t, has)
 	}
 	runServiceTest(t, false, false, fn)
 }
diff --git a/pkg/logservice/store_metadata_test.go b/pkg/logservice/store_metadata_test.go
index 148fa2baa..51eae81db 100644
--- a/pkg/logservice/store_metadata_test.go
+++ b/pkg/logservice/store_metadata_test.go
@@ -26,6 +26,20 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 )
 
+func TestHasMetadataRec(t *testing.T) {
+	cfg := getStoreTestConfig()
+	defer vfs.ReportLeakedFD(cfg.FS, t)
+	cfg.Fill()
+	s := store{cfg: cfg}
+	s.addMetadata(10, 1)
+	has, err := hasMetadataRec(s.cfg.DataDir, logMetadataFilename, 10, 1, s.cfg.FS)
+	require.NoError(t, err)
+	assert.True(t, has)
+	has, err = hasMetadataRec(s.cfg.DataDir, logMetadataFilename, 1, 1, s.cfg.FS)
+	require.NoError(t, err)
+	assert.False(t, has)
+}
+
 func TestAddMetadata(t *testing.T) {
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)