Apply collection start position (#9459)

Signed-off-by: NCongqi Xia <congqi.xia@zilliz.com>

internal/datacoord/server.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/milvus-io/milvus/internal/rootcoord"
 	"github.com/milvus-io/milvus/internal/util/metricsinfo"
 
 	datanodeclient "github.com/milvus-io/milvus/internal/distributed/datanode/client"
@@ -665,6 +666,20 @@ func (s *Server) GetVChanPositions(vchans []vchannel, seekFromStartPosition bool
 			}
 		}
 
+		if seekPosition == nil {
+			coll := s.meta.GetCollection(vchan.CollectionID)
+			if coll != nil {
+				for _, sp := range coll.GetStartPositions() {
+					if sp.GetKey() == rootcoord.ToPhysicalChannel(vchan.DmlChannel) {
+						seekPosition = &internalpb.MsgPosition{
+							ChannelName: vchan.DmlChannel,
+							MsgID:       sp.GetData(),
+						}
+					}
+				}
+			}
+		}
+
 		pairs = append(pairs, &datapb.VchannelInfo{
 			CollectionID:      vchan.CollectionID,
 			ChannelName:       vchan.DmlChannel,