fix watch dml channel remove offline node (#18835)

Signed-off-by: NWei Liu <wei.liu@zilliz.com>
Signed-off-by: NWei Liu <wei.liu@zilliz.com>

internal/proto/query_coord.proto

@@ -183,6 +183,8 @@ message WatchDmChannelsRequest {
   LoadMetaInfo load_meta = 8;
   int64 replicaID = 9;
   map<int64, data.SegmentInfo> segment_infos = 10;
+  // for node down load balance, need to remove offline node in time after every watchDmChannel finish.
+  int64 offlineNodeID = 11;
 }
 
 message SegmentLoadInfo {

internal/querycoord/cluster.go

@@ -301,6 +301,11 @@ func (c *queryNodeCluster) WatchDmChannels(ctx context.Context, nodeID int64, in
 			old, ok := c.clusterMeta.getDmChannel(info.ChannelName)
 			if ok {
 				nodes = append(nodes, old.NodeIds...)
+				removeFromSlice(nodes, in.OfflineNodeID)
+				log.Debug("Remove offline node from dmChannel",
+					zap.String("channel", info.ChannelName),
+					zap.Int64("removedNode", in.OfflineNodeID),
+					zap.Int64s("leftNodes", nodes))
 			}
 
 			dmChannelWatchInfo[index] = &querypb.DmChannelWatchInfo{

internal/querycoord/task.go

@@ -1584,7 +1584,8 @@ func (wdt *watchDmChannelTask) reschedule(ctx context.Context) ([]task, error) {
 				CollectionID: collectionID,
 				PartitionIDs: wdt.GetLoadMeta().GetPartitionIDs(),
 			},
-			ReplicaID: wdt.GetReplicaID(),
+			ReplicaID:     wdt.GetReplicaID(),
+			OfflineNodeID: wdt.OfflineNodeID,
 		}
 		watchDmChannelReqs = append(watchDmChannelReqs, req)
 	}
@@ -2052,7 +2053,8 @@ func (lbt *loadBalanceTask) processNodeDownLoadBalance(ctx context.Context) erro
 							CollectionID: collectionID,
 							PartitionIDs: toRecoverPartitionIDs,
 						},
-						ReplicaID: replica.ReplicaID,
+						ReplicaID:     replica.ReplicaID,
+						OfflineNodeID: nodeID,
 					}
 
 					if collectionInfo.LoadType == querypb.LoadType_LoadPartition {