Repository: matrixorigin/matrixone
Pull request #3893 (2022-07-11 11:00:31)

hakeeper: respect per determined id ranges
**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:**

k8s, hakeeper in bootstrap mode and hakeeper in running mode now all get their own id ranges, when allocating shard/replica id values, they will have to draw from their per determined id ranges.  

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/hakeeper/rsm.go b/pkg/hakeeper/rsm.go
index 929aa37bc..da74e5d19 100644
--- a/pkg/hakeeper/rsm.go
+++ b/pkg/hakeeper/rsm.go
@@ -336,6 +336,13 @@ func (s *stateMachine) handleInitialClusterRequestCmd(cmd []byte) sm.Result {
 		DNShards:  dnShards,
 		LogShards: logShards,
 	}
+
+	// make sure we are not using the ID range assigned to k8s
+	if s.state.NextID > K8SIDRangeStart {
+		panic("too many IDs assigned during initial cluster request")
+	}
+	s.state.NextID = K8SIDRangeEnd
+
 	plog.Infof("initial cluster set, HAKeeper is in BOOTSTRAPPING state")
 	s.state.State = pb.HAKeeperBootstrapping
 	return result
diff --git a/pkg/logservice/hachecker.go b/pkg/logservice/hachecker.go
index f0115ad9f..ebd6e6456 100644
--- a/pkg/logservice/hachecker.go
+++ b/pkg/logservice/hachecker.go
@@ -56,6 +56,11 @@ func (a *idAllocator) Next() (uint64, bool) {
 }
 
 func (a *idAllocator) Set(next uint64, last uint64) {
+	// make sure that this id allocator never emit any id smaller than
+	// K8SIDRangeEnd
+	if next < hakeeper.K8SIDRangeEnd {
+		panic("invalid id allocator range")
+	}
 	a.nextID = next
 	a.lastID = last
 }


Test patch:
diff --git a/pkg/hakeeper/rsm_test.go b/pkg/hakeeper/rsm_test.go
index 823d5eb7b..54c66e6a1 100644
--- a/pkg/hakeeper/rsm_test.go
+++ b/pkg/hakeeper/rsm_test.go
@@ -417,5 +417,5 @@ func TestHandleInitialClusterRequestCmd(t *testing.T) {
 
 	assert.Equal(t, expected, rsm.state.ClusterInfo)
 	assert.Equal(t, pb.HAKeeperBootstrapping, rsm.state.State)
-	assert.Equal(t, uint64(5), rsm.state.NextID)
+	assert.Equal(t, K8SIDRangeEnd, rsm.state.NextID)
 }
diff --git a/pkg/logservice/hachecker_test.go b/pkg/logservice/hachecker_test.go
index 9e1127402..92a0e0e59 100644
--- a/pkg/logservice/hachecker_test.go
+++ b/pkg/logservice/hachecker_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -57,8 +58,22 @@ func TestIDAllocatorCapacity(t *testing.T) {
 
 func TestIDAllocatorSet(t *testing.T) {
 	alloc := idAllocator{nextID: 100, lastID: 200}
-	alloc.Set(200, 300)
-	assert.Equal(t, idAllocator{nextID: 200, lastID: 300}, alloc)
+	alloc.Set(hakeeper.K8SIDRangeEnd, hakeeper.K8SIDRangeEnd+100)
+	expected := idAllocator{
+		nextID: hakeeper.K8SIDRangeEnd,
+		lastID: hakeeper.K8SIDRangeEnd + 100,
+	}
+	assert.Equal(t, expected, alloc)
+}
+
+func TestIDAllocatorRejectInvalidSetInput(t *testing.T) {
+	alloc := idAllocator{nextID: 100, lastID: 200}
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatalf("failed to trigger panic")
+		}
+	}()
+	alloc.Set(300, 400)
 }
 
 func TestIDAllocatorNext(t *testing.T) {
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 4da39b5f9..066347c6c 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -70,9 +70,11 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	store, err := newLogStore(cfg)
 	assert.NoError(t, err)
+	plog.Infof("1")
 	defer func() {
 		assert.NoError(t, store.close())
 	}()
+	plog.Infof("2")
 }
 
 func getTestStore(cfg Config, startLogReplica bool) (*store, error) {