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

fix: data lost when wal balance (#42149)

Browse Source 
issue: #42147

- error of sync task should be returned if error is returned to avoid
checkpoint is push forward.
- fix up node id checker of UpdateChannelCheckpoint in streaming.

Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
Zhen Ye 2025-05-29 17:32:29 +08:00 committed by GitHub
parent 3a74044149
commit c7d6e3f19b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 19 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
internal/datacoord/services.go
View File

@ -1490,7 +1490,14 @@ func (s *Server) UpdateChannelCheckpoint(ctx context.Context, req *datapb.Update
// For compatibility with old client // For compatibility with old client
if req.GetVChannel() != "" && req.GetPosition() != nil { if req.GetVChannel() != "" && req.GetPosition() != nil {
channel := req.GetVChannel() channel := req.GetVChannel()
if !s.channelManager.Match(nodeID, channel) { if streamingutil.IsStreamingServiceEnabled() {
targetID, err := snmanager.StaticStreamingNodeManager.GetLatestWALLocated(ctx, channel)
if err != nil || targetID != nodeID {
err := merr.WrapErrChannelNotFound(channel, fmt.Sprintf("for node %d", nodeID))
log.Warn("failed to get latest wal allocated", zap.Error(err))
return merr.Status(err), nil
}
} else if !s.channelManager.Match(nodeID, channel) {
log.Warn("node is not matched with channel", zap.String("channel", channel), zap.Int64("nodeID", nodeID)) log.Warn("node is not matched with channel", zap.String("channel", channel), zap.Int64("nodeID", nodeID))
return merr.Status(merr.WrapErrChannelNotFound(channel, fmt.Sprintf("from node %d", nodeID))), nil return merr.Status(merr.WrapErrChannelNotFound(channel, fmt.Sprintf("from node %d", nodeID))), nil
} }
@ -1504,11 +1511,19 @@ func (s *Server) UpdateChannelCheckpoint(ctx context.Context, req *datapb.Update
checkpoints := lo.Filter(req.GetChannelCheckpoints(), func(cp *msgpb.MsgPosition, _ int) bool { checkpoints := lo.Filter(req.GetChannelCheckpoints(), func(cp *msgpb.MsgPosition, _ int) bool {
channel := cp.GetChannelName() channel := cp.GetChannelName()
matched := s.channelManager.Match(nodeID, channel) if streamingutil.IsStreamingServiceEnabled() {
if !matched { targetID, err := snmanager.StaticStreamingNodeManager.GetLatestWALLocated(ctx, channel)
if err != nil || targetID != nodeID {
err := merr.WrapErrChannelNotFound(channel, fmt.Sprintf("for node %d", nodeID))
log.Warn("failed to get latest wal allocated", zap.Error(err))
return false
}
return true
} else if !s.channelManager.Match(nodeID, channel) {
log.Warn("node is not matched with channel", zap.String("channel", channel), zap.Int64("nodeID", nodeID)) log.Warn("node is not matched with channel", zap.String("channel", channel), zap.Int64("nodeID", nodeID))
return false
} }
return matched return true
}) })
err := s.meta.UpdateChannelCheckpoints(ctx, checkpoints) err := s.meta.UpdateChannelCheckpoints(ctx, checkpoints)