enhance: Remove legacy cdc/replication (milvus-io#46603)

issue: milvus-io#44123

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
- Core invariant: legacy in-cluster CDC/replication plumbing
(ReplicateMsg types, ReplicateID-based guards and flags) is obsolete —
the system relies on standard msgstream positions, subPos/end-ts
semantics and timetick ordering as the single source of truth for
message ordering and skipping, so replication-specific
channels/types/guards can be removed safely.

- Removed/simplified logic (what and why): removed replication feature
flags and params (ReplicateMsgChannel, TTMsgEnabled,
CollectionReplicateEnable), ReplicateMsg type and its tests, ReplicateID
constants/helpers and MergeProperties hooks, ReplicateConfig and its
propagation (streamPipeline, StreamConfig, dispatcher, target),
replicate-aware dispatcher/pipeline branches, and replicate-mode
pre-checks/timestamp-allocation in proxy tasks — these implemented a
redundant alternate “replicate-mode” pathway that duplicated
position/end-ts and timetick logic.

- Why this does NOT cause data loss or regression (concrete code paths):
no persistence or core write paths were removed — proxy PreExecute flows
(internal/proxy/task_*.go) still perform the same schema/ID/size
validations and then follow the normal non-replicate execution path;
dispatcher and pipeline continue to use position/subPos and
pullback/end-ts in Seek/grouping (pkg/mq/msgdispatcher/dispatcher.go,
internal/util/pipeline/stream_pipeline.go), so skipping and ordering
behavior remains unchanged; timetick emission in rootcoord
(sendMinDdlTsAsTt) is now ungated (no silent suppression), preserving or
increasing timetick delivery rather than removing it.

- PR type and net effect: Enhancement/Refactor — removes deprecated
replication API surface (types, helpers, config, tests) and replication
branches, simplifies public APIs and constructor signatures, and reduces
surface area for future maintenance while keeping DML/DDL persistence,
ordering, and seek semantics intact.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

Author: bigsheeper

configs/milvus.yaml

@@ -841,7 +841,6 @@ msgChannel:
     # Caution: Changing this parameter after using Milvus for a period of time will affect your access to old data.
     # It is recommended to change this parameter before starting Milvus for the first time.
     rootCoordDml: rootcoord-dml
-    replicateMsg: replicate-msg
     # Sub-name prefix of the message channel where the query node publishes time tick messages.
     # The complete channel name prefix is ${msgChannel.chanNamePrefix.cluster}-${msgChannel.chanNamePrefix.queryTimeTick}
     # Caution: Changing this parameter after using Milvus for a period of time will affect your access to old data.
@@ -1024,16 +1023,11 @@ common:
         enabled: false # enable split by average size policy in storage v2
         threshold: 1024 # split by average size policy threshold(in bytes) in storage v2
     useLoonFFI: false
-  # Whether to disable the internal time messaging mechanism for the system.
-  # If disabled (set to false), the system will not allow DML operations, including insertion, deletion, queries, and searches.
-  # This helps Milvus-CDC synchronize incremental data
-  ttMsgEnabled: true
   traceLogMode: 0 # trace request info
   bloomFilterSize: 100000 # bloom filter initial size
   bloomFilterType: BlockedBloomFilter # bloom filter type, support BasicBloomFilter and BlockedBloomFilter
   maxBloomFalsePositive: 0.001 # max false positive rate for bloom filter
   bloomFilterApplyBatchSize: 1000 # batch size when to apply pk to bloom filter
-  collectionReplicateEnable: false # Whether to enable collection replication.
   usePartitionKeyAsClusteringKey: false # if true, do clustering compaction and segment prune on partition key field
   useVectorAsClusteringKey: false # if true, do clustering compaction and segment prune on vector field
   enableVectorClusteringKey: false # if true, enable vector clustering key and vector clustering compaction

internal/distributed/streaming/msgstream_adaptor.go

@@ -129,7 +129,3 @@ func (m *delegatorMsgstreamAdaptor) GetLatestMsgID(channel string) (msgstream.Me
 func (m *delegatorMsgstreamAdaptor) CheckTopicValid(channel string) error {
 	panic("should never be called")
 }
-
-func (m *delegatorMsgstreamAdaptor) ForceEnableProduce(can bool) {
-	panic("should never be called")
-}

internal/distributed/streaming/msgstream_adaptor_test.go

@@ -161,14 +161,4 @@ func TestDelegatorMsgstreamAdaptor(t *testing.T) {
 		}()
 		_ = adaptor.CheckTopicValid("channel1")
 	})
-
-	// Test ForceEnableProduce
-	t.Run("ForceEnableProduce", func(t *testing.T) {
-		defer func() {
-			if r := recover(); r == nil {
-				t.Errorf("ForceEnableProduce should panic but did not")
-			}
-		}()
-		adaptor.ForceEnableProduce(true)
-	})
 }

internal/flushcommon/pipeline/flow_graph_dmstream_input_node.go

@@ -27,7 +27,6 @@ import (
 	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/internal/util/flowgraph"
-	pkgcommon "github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
 	"github.com/milvus-io/milvus/pkg/v2/mq/common"
@@ -76,19 +75,11 @@ func createNewInputFromDispatcher(initCtx context.Context,
 		start = time.Now()
 	)
 
-	replicateID, _ := pkgcommon.GetReplicateID(schema.GetProperties())
-	if replicateID == "" {
-		log.Info("datanode consume without replicateID, try to get replicateID from dbProperties", zap.Any("dbProperties", dbProperties))
-		replicateID, _ = pkgcommon.GetReplicateID(dbProperties)
-	}
-	replicateConfig := msgstream.GetReplicateConfig(replicateID, schema.GetDbName(), schema.GetName())
-
 	if seekPos != nil && len(seekPos.MsgID) != 0 {
 		input, err = dispatcherClient.Register(initCtx, &msgdispatcher.StreamConfig{
-			VChannel:        vchannel,
-			Pos:             seekPos,
-			SubPos:          common.SubscriptionPositionUnknown,
-			ReplicateConfig: replicateConfig,
+			VChannel: vchannel,
+			Pos:      seekPos,
+			SubPos:   common.SubscriptionPositionUnknown,
 		})
 		if err != nil {
 			log.Warn("datanode consume failed after retried", zap.Error(err))
@@ -105,10 +96,9 @@ func createNewInputFromDispatcher(initCtx context.Context,
 	}
 
 	input, err = dispatcherClient.Register(initCtx, &msgdispatcher.StreamConfig{
-		VChannel:        vchannel,
-		Pos:             nil,
-		SubPos:          common.SubscriptionPositionEarliest,
-		ReplicateConfig: replicateConfig,
+		VChannel: vchannel,
+		Pos:      nil,
+		SubPos:   common.SubscriptionPositionEarliest,
 	})
 	if err != nil {
 		log.Warn("datanode consume failed after retried", zap.Error(err))

internal/flushcommon/pipeline/flow_graph_dmstream_input_node_test.go

@@ -62,9 +62,6 @@ func (mm *mockMsgStreamFactory) NewMsgStreamDisposer(ctx context.Context) func([
 
 type mockTtMsgStream struct{}
 
-func (mtm *mockTtMsgStream) SetReplicate(config *msgstream.ReplicateConfig) {
-}
-
 func (mtm *mockTtMsgStream) Close() {}
 
 func (mtm *mockTtMsgStream) Chan() <-chan *msgstream.ConsumeMsgPack {
@@ -107,9 +104,6 @@ func (mtm *mockTtMsgStream) CheckTopicValid(channel string) error {
 	return nil
 }
 
-func (mtm *mockTtMsgStream) ForceEnableProduce(can bool) {
-}
-
 func TestNewDmInputNode(t *testing.T) {
 	assert.Panics(t, func() {
 		newDmInputNode(&nodeConfig{

internal/proxy/meta_cache.go

@@ -99,7 +99,6 @@ type collectionInfo struct {
 	createdUtcTimestamp   uint64
 	consistencyLevel      commonpb.ConsistencyLevel
 	partitionKeyIsolation bool
-	replicateID           string
 	updateTimestamp       uint64
 	collectionTTL         uint64
 	numPartitions         int64
@@ -489,7 +488,6 @@ func (m *MetaCache) update(ctx context.Context, database, collectionName string,
 		m.collInfo[database] = make(map[string]*collectionInfo)
 	}
 
-	replicateID, _ := common.GetReplicateID(collection.Properties)
 	m.collInfo[database][collectionName] = &collectionInfo{
 		collID:                collection.CollectionID,
 		schema:                schemaInfo,
@@ -498,7 +496,6 @@ func (m *MetaCache) update(ctx context.Context, database, collectionName string,
 		createdUtcTimestamp:   collection.CreatedUtcTimestamp,
 		consistencyLevel:      collection.ConsistencyLevel,
 		partitionKeyIsolation: isolation,
-		replicateID:           replicateID,
 		updateTimestamp:       collection.UpdateTimestamp,
 		collectionTTL:         getCollectionTTL(schemaInfo.CollectionSchema.GetProperties()),
 		vChannels:             collection.VirtualChannelNames,

internal/proxy/mock_msgstream_test.go

@@ -10,10 +10,9 @@ import (
 
 type mockMsgStream struct {
 	msgstream.MsgStream
-	asProducer         func([]string)
-	setRepack          func(repackFunc msgstream.RepackFunc)
-	close              func()
-	forceEnableProduce func(bool)
+	asProducer func([]string)
+	setRepack  func(repackFunc msgstream.RepackFunc)
+	close      func()
 }
 
 func (m *mockMsgStream) AsProducer(ctx context.Context, producers []string) {
@@ -34,15 +33,6 @@ func (m *mockMsgStream) Close() {
 	}
 }
 
-func (m *mockMsgStream) ForceEnableProduce(enabled bool) {
-	if m.forceEnableProduce != nil {
-		m.forceEnableProduce(enabled)
-	}
-}
-
-func (m *mockMsgStream) SetReplicate(config *msgstream.ReplicateConfig) {
-}
-
 func newMockMsgStream() *mockMsgStream {
 	return &mockMsgStream{}
 }

internal/proxy/mock_test.go

@@ -314,12 +314,6 @@ func (ms *simpleMockMsgStream) CheckTopicValid(topic string) error {
 	return nil
 }
 
-func (ms *simpleMockMsgStream) ForceEnableProduce(enabled bool) {
-}
-
-func (ms *simpleMockMsgStream) SetReplicate(config *msgstream.ReplicateConfig) {
-}
-
 func newSimpleMockMsgStream() *simpleMockMsgStream {
 	return &simpleMockMsgStream{
 		msgChan:  make(chan *msgstream.ConsumeMsgPack, 1024),

internal/proxy/repack_func.go

@@ -81,14 +81,3 @@ func defaultInsertRepackFunc(
 	}
 	return pack, nil
 }
-
-func replicatePackFunc(
-	tsMsgs []msgstream.TsMsg,
-	hashKeys [][]int32,
-) (map[int32]*msgstream.MsgPack, error) {
-	return map[int32]*msgstream.MsgPack{
-		0: {
-			Msgs: tsMsgs,
-		},
-	}, nil
-}

internal/proxy/task.go

@@ -37,7 +37,6 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/mq/msgstream"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
-	"github.com/milvus-io/milvus/pkg/v2/proto/rootcoordpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/commonpbutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
@@ -1366,26 +1365,6 @@ func (t *alterCollectionTask) PreExecute(ctx context.Context) error {
 				"can not alter partition key isolation mode if the collection already has a vector index. Please drop the index first")
 		}
 	}
-
-	_, ok := common.IsReplicateEnabled(t.Properties)
-	if ok {
-		return merr.WrapErrParameterInvalidMsg("can't set the replicate.id property")
-	}
-	endTS, ok := common.GetReplicateEndTS(t.Properties)
-	if ok && collBasicInfo.replicateID != "" {
-		allocResp, err := t.mixCoord.AllocTimestamp(ctx, &rootcoordpb.AllocTimestampRequest{
-			Count:          1,
-			BlockTimestamp: endTS,
-		})
-		if err = merr.CheckRPCCall(allocResp, err); err != nil {
-			return merr.WrapErrServiceInternal("alloc timestamp failed", err.Error())
-		}
-		if allocResp.GetTimestamp() <= endTS {
-			return merr.WrapErrServiceInternal("alter collection: alloc timestamp failed, timestamp is not greater than endTS",
-				fmt.Sprintf("timestamp = %d, endTS = %d", allocResp.GetTimestamp(), endTS))
-		}
-	}
-
 	return nil
 }