enhance: allow pchannel count increase in ReplicateConfiguration (milvus-io#47792)

## Summary

- Allow pchannel count increase (append-only) in
`validateClusterConsistency` for CU scaling
- Existing pchannels must be preserved at the same positions; decrease
and reorder are still rejected
- Equal pchannel count across clusters is still enforced in
`validateClusterBasic`

## Test plan

- [x] Unit tests pass for `pkg/util/replicateutil/...`
- [x] Verified pchannel increase (append) is accepted
- [x] Verified pchannel decrease is rejected
- [x] Verified pchannel reorder/replace is rejected

issue: milvus-io#47791

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: bigsheeper

internal/cdc/replication/replicatemanager/channel_replicator.go

@@ -181,6 +181,17 @@ func (r *channelReplicator) startConsumeLoop() {
 func (r *channelReplicator) getReplicateCheckpoint() (*utility.ReplicateCheckpoint, error) {
 	logger := log.With(zap.String("key", r.channel.Key), zap.Int64("modRevision", r.channel.ModRevision))
 
+	// For pchannel-increasing tasks, the secondary WAL for new pchannels hasn't received the
+	// AlterReplicateConfig yet, so GetReplicateInfo would fail. Use InitializedCheckpoint directly.
+	if r.channel.Value.GetSkipGetReplicateCheckpoint() {
+		initializedCheckpoint := utility.NewReplicateCheckpointFromProto(r.channel.Value.InitializedCheckpoint)
+		logger.Info("skip get replicate checkpoint for pchannel-increasing task, use initialized checkpoint",
+			zap.Stringer("messageID", initializedCheckpoint.MessageID),
+			zap.Uint64("timeTick", initializedCheckpoint.TimeTick),
+		)
+		return initializedCheckpoint, nil
+	}
+
 	ctx, cancel := context.WithTimeout(r.asyncNotifier.Context(), 30*time.Second)
 	defer cancel()
 

internal/distributed/streaming/replicate_service.go

@@ -120,17 +120,36 @@ func (s replicateService) overwriteReplicateMessage(ctx context.Context, msg mes
 	if sourceCluster == nil {
 		return nil, status.NewReplicateViolation("source cluster %s not found in replicate configuration", rh.ClusterID)
 	}
-	targetVChannel, err := s.getTargetVChannel(sourceCluster, msg.VChannel())
+
+	// For pchannel-increasing AlterReplicateConfig messages, use the NEW config from the
+	// message header to map ALL channels (including newly added ones).
+	// The current config only knows about old pchannels, so both the main vchannel and
+	// broadcast vchannels need the new config for mapping.
+	channelMappingSourceCluster := sourceCluster
+	if msg.MessageType() == message.MessageTypeAlterReplicateConfig {
+		alterMsg := message.MustAsMutableAlterReplicateConfigMessageV2(msg)
+		if alterMsg.Header().GetIsPchannelIncreasing() {
+			newCfg, newCfgErr := replicateutil.NewConfigHelper(s.clusterID, alterMsg.Header().GetReplicateConfiguration())
+			if newCfgErr != nil {
+				return nil, status.NewReplicateViolation("failed to parse new replicate config from message header: %s", newCfgErr.Error())
+			}
+			channelMappingSourceCluster = newCfg.GetCluster(rh.ClusterID)
+			if channelMappingSourceCluster == nil {
+				return nil, status.NewReplicateViolation("source cluster %s not found in new replicate configuration", rh.ClusterID)
+			}
+		}
+	}
+
+	targetVChannel, err := s.getTargetVChannel(channelMappingSourceCluster, msg.VChannel())
 	if err != nil {
 		return nil, err
 	}
 
 	// Get target broadcast vchannels on current cluster that should be written to
 	if bh := msg.BroadcastHeader(); bh != nil {
-		// broadcast header have vchannels, so we need to overwrite it.
 		targetBroadcastVChannels := make([]string, 0, len(bh.VChannels))
 		for _, vchannel := range bh.VChannels {
-			targetBroadcastVChannel, err := s.getTargetVChannel(sourceCluster, vchannel)
+			targetBroadcastVChannel, err := s.getTargetVChannel(channelMappingSourceCluster, vchannel)
 			if err != nil {
 				return nil, status.NewReplicateViolation("failed to get target channel, %s", err.Error())
 			}

internal/distributed/streaming/replicate_service_test.go

@@ -397,6 +397,228 @@ func TestReplicateService_SkipMessageTypes(t *testing.T) {
 	})
 }
 
+func TestReplicateService_AlterConfigPChannelIncreasing(t *testing.T) {
+	// New config adds a 3rd channel (dml_2)
+	newConfig := &commonpb.ReplicateConfiguration{
+		Clusters: []*commonpb.MilvusCluster{
+			{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1", "primary-rootcoord-dml_2"}},
+			{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0", "by-dev-rootcoord-dml_1", "by-dev-rootcoord-dml_2"}},
+		},
+		CrossClusterTopology: []*commonpb.CrossClusterTopology{
+			{SourceClusterId: "primary", TargetClusterId: "by-dev"},
+		},
+	}
+
+	t.Run("with_flag_maps_all_channels", func(t *testing.T) {
+		c := mock_client.NewMockClient(t)
+		as := mock_client.NewMockAssignmentService(t)
+		c.EXPECT().Assignment().Return(as).Maybe()
+
+		h := mock_handler.NewMockHandlerClient(t)
+		p := mock_producer.NewMockProducer(t)
+		p.EXPECT().Append(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, mm message.MutableMessage) (*types.AppendResult, error) {
+			msg := message.MustAsMutableAlterReplicateConfigMessageV2(mm)
+			// With IsPchannelIncreasing flag, all 3 channels (including new one)
+			// should be mapped using the new config from the message header.
+			bh := msg.BroadcastHeader()
+			assert.NotNil(t, bh)
+			assert.Len(t, bh.VChannels, 3, "all channels including new one should be mapped")
+			for _, vchannel := range bh.VChannels {
+				assert.True(t, strings.HasPrefix(vchannel, "by-dev"), "vchannel should be mapped to secondary cluster")
+			}
+			return &types.AppendResult{
+				MessageID: walimplstest.NewTestMessageID(1),
+				TimeTick:  1,
+			}, nil
+		}).Maybe()
+		p.EXPECT().IsAvailable().Return(true).Maybe()
+		p.EXPECT().Available().Return(make(chan struct{})).Maybe()
+		h.EXPECT().CreateProducer(mock.Anything, mock.Anything).Return(p, nil).Maybe()
+
+		// Current (old) config has 2 channels
+		as.EXPECT().GetReplicateConfiguration(mock.Anything).Return(replicateutil.MustNewConfigHelper(
+			"by-dev",
+			&commonpb.ReplicateConfiguration{
+				Clusters: []*commonpb.MilvusCluster{
+					{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1"}},
+					{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0", "by-dev-rootcoord-dml_1"}},
+				},
+				CrossClusterTopology: []*commonpb.CrossClusterTopology{
+					{SourceClusterId: "primary", TargetClusterId: "by-dev"},
+				},
+			},
+		), nil)
+		as.EXPECT().GetLatestAssignments(mock.Anything).Return(nil, errors.New("not needed")).Maybe()
+
+		rs := &replicateService{
+			walAccesserImpl: &walAccesserImpl{
+				lifetime:             typeutil.NewLifetime(),
+				clusterID:            "by-dev",
+				streamingCoordClient: c,
+				handlerClient:        h,
+				producers:            make(map[string]*producer.ResumableProducer),
+			},
+		}
+
+		// Build AlterReplicateConfig broadcast with 3 channels and IsPchannelIncreasing flag
+		replicateMsgs := createReplicateAlterConfigMessages(newConfig,
+			[]string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1", "primary-rootcoord-dml_2"},
+			true)
+
+		for _, msg := range replicateMsgs {
+			_, err := rs.Append(context.Background(), msg)
+			assert.NoError(t, err)
+		}
+	})
+
+	t.Run("without_flag_fails_for_unknown_channel", func(t *testing.T) {
+		c := mock_client.NewMockClient(t)
+		as := mock_client.NewMockAssignmentService(t)
+		c.EXPECT().Assignment().Return(as).Maybe()
+
+		h := mock_handler.NewMockHandlerClient(t)
+
+		// Current (old) config has 2 channels
+		as.EXPECT().GetReplicateConfiguration(mock.Anything).Return(replicateutil.MustNewConfigHelper(
+			"by-dev",
+			&commonpb.ReplicateConfiguration{
+				Clusters: []*commonpb.MilvusCluster{
+					{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1"}},
+					{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0", "by-dev-rootcoord-dml_1"}},
+				},
+				CrossClusterTopology: []*commonpb.CrossClusterTopology{
+					{SourceClusterId: "primary", TargetClusterId: "by-dev"},
+				},
+			},
+		), nil)
+
+		rs := &replicateService{
+			walAccesserImpl: &walAccesserImpl{
+				lifetime:             typeutil.NewLifetime(),
+				clusterID:            "by-dev",
+				streamingCoordClient: c,
+				handlerClient:        h,
+				producers:            make(map[string]*producer.ResumableProducer),
+			},
+		}
+
+		// Build AlterReplicateConfig broadcast with 3 channels but NO IsPchannelIncreasing flag
+		replicateMsgs := createReplicateAlterConfigMessages(newConfig,
+			[]string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1", "primary-rootcoord-dml_2"},
+			false)
+
+		// Should fail because the old config doesn't know about primary-rootcoord-dml_2
+		for _, msg := range replicateMsgs {
+			_, err := rs.Append(context.Background(), msg)
+			assert.Error(t, err)
+			assert.Contains(t, err.Error(), "failed to get target channel")
+		}
+	})
+
+	t.Run("with_flag_invalid_config_in_header", func(t *testing.T) {
+		c := mock_client.NewMockClient(t)
+		as := mock_client.NewMockAssignmentService(t)
+		c.EXPECT().Assignment().Return(as).Maybe()
+
+		h := mock_handler.NewMockHandlerClient(t)
+
+		// Current config has 2 channels
+		as.EXPECT().GetReplicateConfiguration(mock.Anything).Return(replicateutil.MustNewConfigHelper(
+			"by-dev",
+			&commonpb.ReplicateConfiguration{
+				Clusters: []*commonpb.MilvusCluster{
+					{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1"}},
+					{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0", "by-dev-rootcoord-dml_1"}},
+				},
+				CrossClusterTopology: []*commonpb.CrossClusterTopology{
+					{SourceClusterId: "primary", TargetClusterId: "by-dev"},
+				},
+			},
+		), nil)
+
+		rs := &replicateService{
+			walAccesserImpl: &walAccesserImpl{
+				lifetime:             typeutil.NewLifetime(),
+				clusterID:            "by-dev",
+				streamingCoordClient: c,
+				handlerClient:        h,
+				producers:            make(map[string]*producer.ResumableProducer),
+			},
+		}
+
+		// Build with IsPchannelIncreasing flag but an invalid config (no topology, multiple primary => error)
+		invalidConfig := &commonpb.ReplicateConfiguration{
+			Clusters: []*commonpb.MilvusCluster{
+				{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0"}},
+				{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0"}},
+			},
+			// Missing CrossClusterTopology => both clusters are "primary" => primaryCount != 1
+		}
+		replicateMsgs := createReplicateAlterConfigMessages(invalidConfig,
+			[]string{"primary-rootcoord-dml_0"},
+			true)
+
+		for _, msg := range replicateMsgs {
+			_, err := rs.Append(context.Background(), msg)
+			assert.Error(t, err)
+			assert.Contains(t, err.Error(), "failed to parse new replicate config")
+		}
+	})
+
+	t.Run("with_flag_source_cluster_missing_in_new_config", func(t *testing.T) {
+		c := mock_client.NewMockClient(t)
+		as := mock_client.NewMockAssignmentService(t)
+		c.EXPECT().Assignment().Return(as).Maybe()
+
+		h := mock_handler.NewMockHandlerClient(t)
+
+		// Current config has 2 channels
+		as.EXPECT().GetReplicateConfiguration(mock.Anything).Return(replicateutil.MustNewConfigHelper(
+			"by-dev",
+			&commonpb.ReplicateConfiguration{
+				Clusters: []*commonpb.MilvusCluster{
+					{ClusterId: "primary", Pchannels: []string{"primary-rootcoord-dml_0", "primary-rootcoord-dml_1"}},
+					{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0", "by-dev-rootcoord-dml_1"}},
+				},
+				CrossClusterTopology: []*commonpb.CrossClusterTopology{
+					{SourceClusterId: "primary", TargetClusterId: "by-dev"},
+				},
+			},
+		), nil)
+
+		rs := &replicateService{
+			walAccesserImpl: &walAccesserImpl{
+				lifetime:             typeutil.NewLifetime(),
+				clusterID:            "by-dev",
+				streamingCoordClient: c,
+				handlerClient:        h,
+				producers:            make(map[string]*producer.ResumableProducer),
+			},
+		}
+
+		// Build with IsPchannelIncreasing flag but the new config uses "other-cluster" instead of "primary"
+		// The replicate header has ClusterID="primary", but the new config doesn't contain "primary"
+		missingSourceConfig := &commonpb.ReplicateConfiguration{
+			Clusters: []*commonpb.MilvusCluster{
+				{ClusterId: "other-cluster", Pchannels: []string{"other-rootcoord-dml_0"}},
+				{ClusterId: "by-dev", Pchannels: []string{"by-dev-rootcoord-dml_0"}},
+			},
+			CrossClusterTopology: []*commonpb.CrossClusterTopology{
+				{SourceClusterId: "other-cluster", TargetClusterId: "by-dev"},
+			},
+		}
+		replicateMsgs := createReplicateAlterConfigMessages(missingSourceConfig,
+			[]string{"primary-rootcoord-dml_0"},
+			true)
+
+		for _, msg := range replicateMsgs {
+			_, err := rs.Append(context.Background(), msg)
+			assert.Error(t, err)
+			assert.Contains(t, err.Error(), "source cluster primary not found in new replicate configuration")
+		}
+	})
+}
+
 func TestBuildSkipMessageTypes(t *testing.T) {
 	t.Run("normal", func(t *testing.T) {
 		m := buildSkipMessageTypes([]string{"AlterResourceGroup", "DropResourceGroup"})
@@ -457,6 +679,26 @@ func broadcastMsgToReplicateMsgs(broadcastMsg message.BroadcastMutableMessage) [
 	return replicateMsgs
 }
 
+func createReplicateAlterConfigMessages(newConfig *commonpb.ReplicateConfiguration, broadcastChannels []string, isPchannelIncreasing bool) []message.ReplicateMutableMessage {
+	alterMsg := message.NewAlterReplicateConfigMessageBuilderV2().
+		WithHeader(&message.AlterReplicateConfigMessageHeader{
+			ReplicateConfiguration: newConfig,
+			IsPchannelIncreasing:   isPchannelIncreasing,
+		}).
+		WithBody(&message.AlterReplicateConfigMessageBody{}).
+		WithBroadcast(broadcastChannels).
+		MustBuildBroadcast()
+	msgs := alterMsg.WithBroadcastID(200).SplitIntoMutableMessage()
+	replicateMsgs := make([]message.ReplicateMutableMessage, 0, len(msgs))
+	for _, msg := range msgs {
+		immutableMsg := msg.WithLastConfirmedUseMessageID().WithTimeTick(1).IntoImmutableMessage(pulsar2.NewPulsarID(
+			pulsar.NewMessageID(1, 2, 3, 4),
+		))
+		replicateMsgs = append(replicateMsgs, message.MustNewReplicateMessage("primary", immutableMsg.IntoImmutableMessageProto()))
+	}
+	return replicateMsgs
+}
+
 func createReplicateCreateCollectionMessages() []message.ReplicateMutableMessage {
 	schema := &schemapb.CollectionSchema{
 		Fields: []*schemapb.FieldSchema{

internal/streamingcoord/server/balancer/channel/manager.go

@@ -637,14 +637,36 @@ func (cm *ChannelManager) getNewIncomingTask(newConfig *replicateutil.ConfigHelp
 	}
 	incomingReplicatingTasks := make([]*streamingpb.ReplicatePChannelMeta, 0, len(incoming.TargetClusters()))
 	for _, targetCluster := range incoming.TargetClusters() {
-		if current != nil && current.TargetCluster(targetCluster.GetClusterId()) != nil {
-			// target already exists, skip it.
-			continue
+		// Determine which pchannels are new and need CDC tasks.
+		// If the target cluster already exists, only create tasks for newly appended pchannels.
+		newPchannels := targetCluster.GetPchannels()
+		skipGetReplicateCheckpoint := false
+		if current != nil {
+			if currentTarget := current.TargetCluster(targetCluster.GetClusterId()); currentTarget != nil {
+				existingCount := len(currentTarget.GetPchannels())
+				if existingCount >= len(newPchannels) {
+					// No new pchannels, skip this target cluster.
+					continue
+				}
+				// Only process newly appended pchannels (validator ensures existing pchannels are preserved at same positions).
+				newPchannels = newPchannels[existingCount:]
+				// For pchannel-increasing tasks, the secondary WAL for new pchannels hasn't received
+				// the AlterReplicateConfig yet, so GetReplicateInfo would fail. Skip it and use
+				// InitializedCheckpoint directly. The secondary filters out duplicates on restart.
+				skipGetReplicateCheckpoint = true
+			}
 		}
-		// TODO: support add new pchannels into existing clusters.
-		for _, pchannel := range targetCluster.GetPchannels() {
+		for _, pchannel := range newPchannels {
 			sourceClusterID := targetCluster.SourceCluster().ClusterId
 			sourcePChannel := targetCluster.MustGetSourceChannel(pchannel)
+			checkpointTimeTick := appendResults[sourcePChannel].TimeTick
+			if skipGetReplicateCheckpoint {
+				// For pchannel-increasing tasks, the CDC scanner uses DeliverFilterTimeTickGT
+				// (strictly greater than). Subtract 1 so the AlterReplicateConfig message itself
+				// (whose TimeTick == appendResults.TimeTick) is included in the scan.
+				// The secondary needs this message on ALL pchannels for the broadcast to complete.
+				checkpointTimeTick--
+			}
 			incomingReplicatingTasks = append(incomingReplicatingTasks, &streamingpb.ReplicatePChannelMeta{
 				SourceChannelName: sourcePChannel,
 				TargetChannelName: pchannel,
@@ -658,8 +680,9 @@ func (cm *ChannelManager) getNewIncomingTask(newConfig *replicateutil.ConfigHelp
 					ClusterId: sourceClusterID,
 					Pchannel:  sourcePChannel,
 					MessageId: appendResults[sourcePChannel].LastConfirmedMessageID.IntoProto(),
-					TimeTick:  appendResults[sourcePChannel].TimeTick,
+					TimeTick:  checkpointTimeTick,
 				},
+				SkipGetReplicateCheckpoint: skipGetReplicateCheckpoint,
 			})
 		}
 	}