Repository: milvus-io/milvus
File: internal/distributed/streaming/replicate_service_test.go (Go)

Code before:
			WithBroadcast([]string{"primary-rootcoord-dml_0_cchannel"}).
			MustBuildBroadcast()
		msgs := broadcastMsgToReplicateMsgs(broadcastMsg)
		for _, msg := range msgs {
			_, err := rs.Append(context.Background(), msg)
			assert.Error(t, err)
			se := status.AsStreamingError(err)
			assert.NotNil(t, se)
			assert.True(t, se.IsIgnoredOperation())
		}
	})

	t.Run("non_skipped_message_passthrough", func(t *testing.T) {
		old := paramtable.Get().StreamingCfg.ReplicationSkipMessageTypes.SwapTempValue("AlterResourceGroup,DropResourceGroup")
		defer paramtable.Get().StreamingCfg.ReplicationSkipMessageTypes.SwapTempValue(old)

		rs := newReplicateService(t)
		replicateMsgs := createReplicateCreateCollectionMessages()
		for _, msg := range replicateMsgs {
			_, err := rs.Append(context.Background(), msg)
			assert.NoError(t, err)
		}
	})

	t.Run("empty_skip_list", func(t *testing.T) {
		old := paramtable.Get().StreamingCfg.ReplicationSkipMessageTypes.SwapTempValue("")
		defer paramtable.Get().StreamingCfg.ReplicationSkipMessageTypes.SwapTempValue(old)

		rs := newReplicateService(t)
		replicateMsgs := createReplicateCreateCollectionMessages()
		for _, msg := range replicateMsgs {
			_, err := rs.Append(context.Background(), msg)
			assert.NoError(t, err)
		}
	})
}

func broadcastMsgToReplicateMsgs(broadcastMsg message.BroadcastMutableMessage) []message.ReplicateMutableMessage {
	msgs := broadcastMsg.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{

Review comment (suggestion):
  old := paramtable.Get().StreamingCfg.ReplicationSkipMessageTypes.SwapTempValue("")

this is actually change temp value to default, not empty

Code after:
		rs := newReplicateServiceForAlterLoadConfig(t, func(t *testing.T, mm message.MutableMessage) {
			alterLoadConfigMsg := message.MustAsMutableAlterLoadConfigMessageV2(mm)
			assert.False(t, alterLoadConfigMsg.Header().GetUseLocalReplicaConfig(),
				"replicated AlterLoadConfig should have UseLocalReplicaConfig=false when config is disabled")
		})

		replicateMsgs := createReplicateAlterLoadConfigMessages()
		for _, msg := range replicateMsgs {
			_, err := rs.Append(context.Background(), msg)
			assert.NoError(t, err)
		}
	})
}

func TestReplicateService_SkipMessageTypes(t *testing.T) {
	newReplicateService := func(t *testing.T, skipTypes map[string]struct{}) *replicateService {
		c := mock_client.NewMockClient(t)
		as := mock_client.NewMockAssignmentService(t)
		c.EXPECT().Assignment().Return(as).Maybe()
		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).Maybe()

		h := mock_handler.NewMockHandlerClient(t)
		p := mock_producer.NewMockProducer(t)
		p.EXPECT().Append(mock.Anything, mock.Anything).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()

		return &replicateService{
			walAccesserImpl: &walAccesserImpl{
				lifetime:             typeutil.NewLifetime(),
				clusterID:            "by-dev",
				streamingCoordClient: c,
				handlerClient:        h,
				producers:            make(map[string]*producer.ResumableProducer),
			},
			skipMessageTypes: skipTypes,