fix: Verify sync task target segment and retry if not match

Makefile

@@ -448,6 +448,7 @@ generate-mockery-datanode: getdeps
 	$(INSTALL_PATH)/mockery --name=SyncManager --dir=$(PWD)/internal/datanode/syncmgr --output=$(PWD)/internal/datanode/syncmgr --filename=mock_sync_manager.go --with-expecter --structname=MockSyncManager --outpkg=syncmgr --inpackage
 	$(INSTALL_PATH)/mockery --name=MetaWriter --dir=$(PWD)/internal/datanode/syncmgr --output=$(PWD)/internal/datanode/syncmgr --filename=mock_meta_writer.go --with-expecter --structname=MockMetaWriter --outpkg=syncmgr --inpackage
 	$(INSTALL_PATH)/mockery --name=Serializer --dir=$(PWD)/internal/datanode/syncmgr --output=$(PWD)/internal/datanode/syncmgr --filename=mock_serializer.go --with-expecter --structname=MockSerializer --outpkg=syncmgr --inpackage
+	$(INSTALL_PATH)/mockery --name=Task --dir=$(PWD)/internal/datanode/syncmgr --output=$(PWD)/internal/datanode/syncmgr --filename=mock_task.go --with-expecter --structname=MockTask --outpkg=syncmgr --inpackage
 	$(INSTALL_PATH)/mockery --name=WriteBuffer --dir=$(PWD)/internal/datanode/writebuffer --output=$(PWD)/internal/datanode/writebuffer --filename=mock_write_buffer.go --with-expecter --structname=MockWriteBuffer --outpkg=writebuffer --inpackage
 	$(INSTALL_PATH)/mockery --name=BufferManager --dir=$(PWD)/internal/datanode/writebuffer --output=$(PWD)/internal/datanode/writebuffer --filename=mock_mananger.go --with-expecter --structname=MockBufferManager --outpkg=writebuffer --inpackage
 	$(INSTALL_PATH)/mockery --name=BinlogIO --dir=$(PWD)/internal/datanode/io --output=$(PWD)/internal/datanode/io --filename=mock_binlogio.go --with-expecter --structname=MockBinlogIO --outpkg=io --inpackage

internal/datanode/syncmgr/key_lock_dispatcher.go

@@ -8,6 +8,7 @@ import (
 
 type Task interface {
 	SegmentID() int64
+	CalcTargetSegment() (int64, error)
 	Checkpoint() *msgpb.MsgPosition
 	StartPosition() *msgpb.MsgPosition
 	ChannelName() string

internal/datanode/syncmgr/key_lock_dispatcher_test.go

@@ -6,19 +6,22 @@ import (
 
 	"github.com/stretchr/testify/suite"
 	"go.uber.org/atomic"
-
-	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
 )
 
+/*
 type mockTask struct {
-	ch  chan struct{}
-	err error
+	targetID int64
+	ch       chan struct{}
+	err      error
 }
 
 func (t *mockTask) done() {
 	close(t.ch)
 }
 
+func (t *mockTask) CalcTargetSegment() (int64, error) {
+	return t.targetID, t.err
+}
 func (t *mockTask) SegmentID() int64                  { panic("no implementation") }
 func (t *mockTask) Checkpoint() *msgpb.MsgPosition    { panic("no implementation") }
 func (t *mockTask) StartPosition() *msgpb.MsgPosition { panic("no implementation") }
@@ -34,7 +37,7 @@ func newMockTask(err error) *mockTask {
 		err: err,
 		ch:  make(chan struct{}),
 	}
-}
+}*/
 
 type KeyLockDispatcherSuite struct {
 	suite.Suite
@@ -43,46 +46,56 @@ type KeyLockDispatcherSuite struct {
 func (s *KeyLockDispatcherSuite) TestKeyLock() {
 	d := newKeyLockDispatcher[int64](2)
 
-	t1 := newMockTask(nil)
-	t2 := newMockTask(nil)
+	done := make(chan struct{})
+	t1 := NewMockTask(s.T())
+	t1.EXPECT().Run().Run(func() {
+		<-done
+	}).Return(nil)
+	t2 := NewMockTask(s.T())
+	t2.EXPECT().Run().Return(nil)
 
 	sig := atomic.NewBool(false)
 
 	d.Submit(1, t1)
 
 	go func() {
-		defer t2.done()
 		d.Submit(1, t2)
 
 		sig.Store(true)
 	}()
 
 	s.False(sig.Load(), "task 2 will never be submit before task 1 done")
 
-	t1.done()
+	close(done)
 
 	s.Eventually(sig.Load, time.Second, time.Millisecond*100)
 }
 
 func (s *KeyLockDispatcherSuite) TestCap() {
 	d := newKeyLockDispatcher[int64](1)
 
-	t1 := newMockTask(nil)
-	t2 := newMockTask(nil)
+	t1 := NewMockTask(s.T())
+	t2 := NewMockTask(s.T())
+
+	done := make(chan struct{})
+	t1.EXPECT().Run().Run(func() {
+		<-done
+	}).Return(nil)
+	t2.EXPECT().Run().Return(nil)
 	sig := atomic.NewBool(false)
 
 	d.Submit(1, t1)
 
 	go func() {
-		defer t2.done()
+		// defer t2.done()
 		d.Submit(2, t2)
 
 		sig.Store(true)
 	}()
 
 	s.False(sig.Load(), "task 2 will never be submit before task 1 done")
 
-	t1.done()
+	close(done)
 
 	s.Eventually(sig.Load, time.Second, time.Millisecond*100)
 }