admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-07 17:48:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: datacoord stop get stuck After upgrading from 2.5 to 2.6 (#42674)

Browse Source 
datacoord stop get stuck After upgrading from 2.5 to 2.6
issue:https://github.com/milvus-io/milvus/issues/42656

Signed-off-by: Xianhui.Lin <xianhui.lin@zilliz.com>
This commit is contained in:
Xianhui Lin 2025-06-12 16:56:36 +08:00 committed by GitHub
parent 911a8df17c
commit 98067f5fc6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 21 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/datacoord/compaction_policy_single.go
View File

@ -98,7 +98,7 @@ func (policy *singleCompactionPolicy) triggerOneCollection(ctx context.Context,
views := make([]CompactionView, 0) views := make([]CompactionView, 0)
for _, group := range partSegments { for _, group := range partSegments {
if Params.DataCoordCfg.IndexBasedCompaction.GetAsBool() { if Params.DataCoordCfg.IndexBasedCompaction.GetAsBool() {
group.segments = FilterInIndexedSegments(policy.handler, policy.meta, false, group.segments...) group.segments = FilterInIndexedSegments(ctx, policy.handler, policy.meta, false, group.segments...)
} }
collectionTTL, err := getCollectionTTL(collection.Properties) collectionTTL, err := getCollectionTTL(collection.Properties)

2
internal/datacoord/compaction_trigger.go
View File

@ -356,7 +356,7 @@ func (t *compactionTrigger) handleSignal(signal *compactionSignal) error {
} }
if Params.DataCoordCfg.IndexBasedCompaction.GetAsBool() { if Params.DataCoordCfg.IndexBasedCompaction.GetAsBool() {
group.segments = FilterInIndexedSegments(t.handler, t.meta, signal.isForce, group.segments...) group.segments = FilterInIndexedSegments(context.Background(), t.handler, t.meta, signal.isForce, group.segments...)
} }
coll, err := t.getCollection(group.collectionID) coll, err := t.getCollection(group.collectionID)

13
internal/datacoord/garbage_collector.go
View File

@ -442,7 +442,10 @@ func (gc *garbageCollector) recycleDroppedSegments(ctx context.Context) {
droppedCompactTo[to] = struct{}{} droppedCompactTo[to] = struct{}{}
} }
} }
indexedSegments := FilterInIndexedSegments(gc.handler, gc.meta, false, lo.Keys(droppedCompactTo)...) indexedSegments := FilterInIndexedSegments(ctx, gc.handler, gc.meta, false, lo.Keys(droppedCompactTo)...)
if ctx.Err() != nil {
return
}
indexedSet := make(typeutil.UniqueSet) indexedSet := make(typeutil.UniqueSet)
for _, segment := range indexedSegments { for _, segment := range indexedSegments {
indexedSet.Insert(segment.GetID()) indexedSet.Insert(segment.GetID())
@ -525,9 +528,13 @@ func (gc *garbageCollector) recycleChannelCPMeta(ctx context.Context) {
_, ok := collectionID2GcStatus[collectionID] _, ok := collectionID2GcStatus[collectionID]
if !ok { if !ok {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) if ctx.Err() != nil {
// process canceled, stop.
return
}
timeoutCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel() defer cancel()
has, err := gc.option.broker.HasCollection(ctx, collectionID) has, err := gc.option.broker.HasCollection(timeoutCtx, collectionID)
if err == nil && !has { if err == nil && !has {
collectionID2GcStatus[collectionID] = gc.meta.catalog.GcConfirm(ctx, collectionID, -1) collectionID2GcStatus[collectionID] = gc.meta.catalog.GcConfirm(ctx, collectionID, -1)
} else { } else {

2
internal/datacoord/handler.go
View File

@ -142,7 +142,7 @@ func (h *ServerHandler) GetQueryVChanPositions(channel RWChannel, partitionIDs .
segments := h.s.meta.GetRealSegmentsForChannel(channel.GetName()) segments := h.s.meta.GetRealSegmentsForChannel(channel.GetName())
validSegmentInfos := make(map[int64]*SegmentInfo) validSegmentInfos := make(map[int64]*SegmentInfo)
indexedSegments := FilterInIndexedSegments(h, h.s.meta, false, segments...) indexedSegments := FilterInIndexedSegments(context.Background(), h, h.s.meta, false, segments...)
indexed := typeutil.NewUniqueSet(lo.Map(indexedSegments, func(segment *SegmentInfo, _ int) int64 { return segment.GetID() })...) indexed := typeutil.NewUniqueSet(lo.Map(indexedSegments, func(segment *SegmentInfo, _ int) int64 { return segment.GetID() })...)
for _, s := range segments { for _, s := range segments {

11
internal/datacoord/util.go
View File

@ -72,11 +72,15 @@ func VerifyResponse(response interface{}, err error) error {
} }
} }
func FilterInIndexedSegments(handler Handler, mt *meta, skipNoIndexCollection bool, segments ...*SegmentInfo) []*SegmentInfo { func FilterInIndexedSegments(ctx context.Context, handler Handler, mt *meta, skipNoIndexCollection bool, segments ...*SegmentInfo) []*SegmentInfo {
if len(segments) == 0 { if len(segments) == 0 {
return nil return nil
} }
if ctx.Err() != nil {
return nil
}
collectionSegments := lo.GroupBy(segments, func(segment *SegmentInfo) int64 { collectionSegments := lo.GroupBy(segments, func(segment *SegmentInfo) int64 {
return segment.GetCollectionID() return segment.GetCollectionID()
}) })
@ -89,8 +93,9 @@ func FilterInIndexedSegments(handler Handler, mt *meta, skipNoIndexCollection bo
continue continue
} }
ctx, cancel := context.WithTimeout(context.Background(), time.Second*2) timeoutCtx, cancel := context.WithTimeout(ctx, time.Second*2)
coll, err := handler.GetCollection(ctx, collection)
coll, err := handler.GetCollection(timeoutCtx, collection)
cancel() cancel()
if err != nil { if err != nil {
log.Warn("failed to get collection schema", zap.Error(err)) log.Warn("failed to get collection schema", zap.Error(err))