fix: Prevent dispatcher merging if curTs is 0 (milvus-io#34562) (milv…

…us-io#34563)

When the main dispatcher has not yet consumed data, curTs is 0. During
this time, merging dispatchers should not be allowed; otherwise, the
data of the solo dispatcher will be skipped.

issue: milvus-io#34255

pr: milvus-io#34562

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

pkg/mq/msgdispatcher/manager.go

@@ -182,7 +182,7 @@ func (c *dispatcherManager) tryMerge() {
 
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if c.mainDispatcher == nil {
+	if c.mainDispatcher == nil || c.mainDispatcher.CurTs() == 0 {
 		return
 	}
 	candidates := make(map[string]struct{})

pkg/mq/msgdispatcher/manager_test.go

@@ -72,6 +72,12 @@ func TestManager(t *testing.T) {
 		_, err = c.Add(ctx, "mock_vchannel_2", nil, mqwrapper.SubscriptionPositionUnknown)
 		assert.NoError(t, err)
 		assert.Equal(t, 3, c.Num())
+		c.(*dispatcherManager).mainDispatcher.curTs.Store(1000)
+		c.(*dispatcherManager).mu.RLock()
+		for _, d := range c.(*dispatcherManager).soloDispatchers {
+			d.curTs.Store(1000)
+		}
+		c.(*dispatcherManager).mu.RUnlock()
 
 		c.(*dispatcherManager).tryMerge()
 		assert.Equal(t, 1, c.Num())
@@ -97,6 +103,12 @@ func TestManager(t *testing.T) {
 		_, err = c.Add(ctx, "mock_vchannel_2", nil, mqwrapper.SubscriptionPositionUnknown)
 		assert.NoError(t, err)
 		assert.Equal(t, 3, c.Num())
+		c.(*dispatcherManager).mainDispatcher.curTs.Store(1000)
+		c.(*dispatcherManager).mu.RLock()
+		for _, d := range c.(*dispatcherManager).soloDispatchers {
+			d.curTs.Store(1000)
+		}
+		c.(*dispatcherManager).mu.RUnlock()
 
 		checkIntervalK := paramtable.Get().MQCfg.MergeCheckInterval.Key
 		paramtable.Get().Save(checkIntervalK, "0.01")