package util import ( "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb" "github.com/milvus-io/milvus/pkg/v3/streaming/util/message" "github.com/milvus-io/milvus/pkg/v3/util/paramtable" "github.com/milvus-io/milvus/pkg/v3/util/replicateutil" ) func IsReplicationRemovedByAlterReplicateConfigMessage(msg message.ImmutableMessage, replicateInfo *streamingpb.ReplicatePChannelMeta) (replicationRemoved bool) { prcMsg := message.MustAsImmutableAlterReplicateConfigMessageV2(msg) header := prcMsg.Header() // Check ignore field - if true, this message should be ignored // This is used for incomplete switchover messages that should be ignored after force promote if header.Ignore { return false } replicateConfig := header.ReplicateConfiguration currentClusterID := paramtable.Get().CommonCfg.ClusterPrefix.GetValue() currentCluster := replicateutil.MustNewConfigHelper(currentClusterID, replicateConfig).GetCurrentCluster() _, err := currentCluster.GetTargetChannel(replicateInfo.GetSourceChannelName(), replicateInfo.GetTargetCluster().GetClusterId()) if err != nil { // Cannot find the target channel, it means that the `current->target` topology edge is removed, // it means that the replication is removed. return true } return false }