Repository: matrixorigin/matrixone
Pull request #3358 (2022-06-24 11:03:48)

logservice: added the ability to add replica
**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:**

#3330 resolved the issue of missing target UUID. Added LogService ability to actually add replica.  

**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
index 7cd49c7d1..b2ff4b51e 100644
--- a/pkg/logservice/commands.go
+++ b/pkg/logservice/commands.go
@@ -37,7 +37,13 @@ func (s *Service) handleCommands(cmds []hapb.ScheduleCommand) {
 }
 
 func (s *Service) handleAddReplica(cmd hapb.ScheduleCommand) {
-	panic("not implemented due to the lack of target UUID in ScheduleCommand")
+	shardID := cmd.ConfigChange.Replica.ShardID
+	replicaID := cmd.ConfigChange.Replica.ReplicaID
+	epoch := cmd.ConfigChange.Replica.Epoch
+	target := cmd.ConfigChange.Replica.UUID
+	if err := s.store.addReplica(shardID, replicaID, target, epoch); err != nil {
+		plog.Errorf("failed to add replica %v", err)
+	}
 }
 
 func (s *Service) handleRemoveReplica(cmd hapb.ScheduleCommand) {
@@ -54,7 +60,7 @@ func (s *Service) handleStartReplica(cmd hapb.ScheduleCommand) {
 	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)
+			plog.Errorf("failed to start HAKeeper replica %v", err)
 		}
 	} else {
 		if err := s.store.StartReplica(shardID, replicaID, nil, false); err != nil {


Test patch:
diff --git a/pkg/logservice/commands_test.go b/pkg/logservice/commands_test.go
index 1020b7d69..900c86101 100644
--- a/pkg/logservice/commands_test.go
+++ b/pkg/logservice/commands_test.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -69,6 +70,34 @@ func TestHandleStopReplica(t *testing.T) {
 	runServiceTest(t, fn)
 }
 
+func TestHandleAddReplica(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.AddReplica,
+			Replica: hapb.Replica{
+				UUID:      uuid.New().String(),
+				ShardID:   1,
+				ReplicaID: 3,
+				Epoch:     2,
+			},
+		},
+	}
+	service1.handleCommands([]hapb.ScheduleCommand{cmd})
+	count, ok := checkReplicaCount(store1, 1)
+	require.True(t, ok)
+	assert.Equal(t, 3, count)
+}
+
 func TestHandleRemoveReplica(t *testing.T) {
 	store1, store2, err := getTestStores()
 	require.NoError(t, err)
@@ -86,9 +115,22 @@ func TestHandleRemoveReplica(t *testing.T) {
 			Replica: hapb.Replica{
 				ShardID:   1,
 				ReplicaID: 2,
+				Epoch:     2,
 			},
 		},
 	}
-	// TODO: add some checks
 	service1.handleCommands([]hapb.ScheduleCommand{cmd})
+	count, ok := checkReplicaCount(store1, 1)
+	require.True(t, ok)
+	assert.Equal(t, 1, count)
+}
+
+func checkReplicaCount(s *store, shardID uint64) (int, bool) {
+	hb := s.getHeartbeatMessage()
+	for _, info := range hb.Replicas {
+		if info.ShardID == shardID {
+			return len(info.Replicas), true
+		}
+	}
+	return 0, false
 }