package impl import ( "context" "testing" "time" "github.com/AlekSi/pointer" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" appconfig "github.com/tidbcloud/dataflow-service/app/config" config_manager_mock "github.com/tidbcloud/dataflow-service/app/models/config_manager/mock" "github.com/tidbcloud/dataflow-service/app/models/domain" "github.com/tidbcloud/dataflow-service/app/repo/entity" repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock" regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api" regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock" ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1" "github.com/tidbcloud/pkgv2/bizerr" "github.com/tidbcloud/pkgv2/clusterproxy" "github.com/tidbcloud/pkgv2/db/mock_db" "go.uber.org/mock/gomock" "google.golang.org/protobuf/encoding/protojson" ) type changefeedModelRegionalImplTestSuite struct { suite.Suite mockChangefeedCli *regionalapi_mock.ChangefeedClient mockTiCDCClusterCli *regionalapi_mock.TiCDCClusterClient mockTiCDCClusterRepo *repo_mock.TiCDCClusterRepo mockChangefeedRepo *repo_mock.TiCDCChangefeedRepo mockConfigManager *config_manager_mock.ChangefeedConfigManager model *changefeedModelRegionalImpl } func (s *changefeedModelRegionalImplTestSuite) SetupSuite() { s.mockChangefeedCli = regionalapi_mock.NewChangefeedClient(s.T()) s.mockTiCDCClusterCli = regionalapi_mock.NewTiCDCClusterClient(s.T()) s.mockTiCDCClusterRepo = repo_mock.NewTiCDCClusterRepo(s.T()) s.mockChangefeedRepo = repo_mock.NewTiCDCChangefeedRepo(s.T()) s.mockConfigManager = config_manager_mock.NewChangefeedConfigManager(s.T()) s.model = &changefeedModelRegionalImpl{ changefeedCli: s.mockChangefeedCli, cdcClusterCli: s.mockTiCDCClusterCli, ticdcClusterRepo: s.mockTiCDCClusterRepo, changefeedRepo: s.mockChangefeedRepo, configManager: s.mockConfigManager, } } func (s *changefeedModelRegionalImplTestSuite) TestOverrideChangefeedURIConfig() { testcases := []struct { srcUriConfig domain.TiCDCChangefeedURIConfig uriConfigStr string err bool expectedUriConfig domain.TiCDCChangefeedURIConfig }{ { srcUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ Endpoint: pointer.ToString("mysql:3306"), User: pointer.ToString("test"), Password: pointer.ToString("test"), }, uriConfigStr: `{}`, err: false, expectedUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ Endpoint: pointer.ToString("mysql:3306"), User: pointer.ToString("test"), Password: pointer.ToString("test"), }, }, { srcUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", Compression: pointer.ToString("gzip"), }, uriConfigStr: `{}`, err: false, expectedUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", Compression: pointer.ToString("gzip"), }, }, { srcUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ Endpoint: pointer.ToString("mysql:3306"), User: pointer.ToString("test"), Password: pointer.ToString("test"), }, uriConfigStr: `{"partition-num":5}`, err: true, }, { srcUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ Endpoint: pointer.ToString("mysql:3306"), User: pointer.ToString("test"), Password: pointer.ToString("test"), }, uriConfigStr: `{"user":"test2","max-txn-row":1000}`, err: false, expectedUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ Endpoint: pointer.ToString("mysql:3306"), User: pointer.ToString("test"), Password: pointer.ToString("test"), MaxTxnRow: pointer.ToInt(1000), }, }, { srcUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{}, uriConfigStr: `{"user":"test2","max-txn-row":1000}`, err: false, expectedUriConfig: &domain.TiCDCChangefeedMySQLURIConfigData{ User: pointer.ToString("test2"), MaxTxnRow: pointer.ToInt(1000), }, }, } for _, tc := range testcases { uriConfig, err := s.model.overrideChangefeedURIConfig(tc.srcUriConfig, tc.uriConfigStr) if tc.err { s.Error(err) } else { s.NoError(err) s.Equal(tc.expectedUriConfig, uriConfig) } } } func (s *changefeedModelRegionalImplTestSuite) TestOverrideChangefeedReplicaConfig() { testcases := []struct { srcReplicaConfig *ticdc_v1.ReplicaConfig replicaConfigStr string err bool expectedReplicaConfig *ticdc_v1.ReplicaConfig }{ { srcReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, replicaConfigStr: `{}`, err: false, expectedReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, }, { srcReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, replicaConfigStr: `{"abcde":"123"}`, err: true, }, { srcReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, replicaConfigStr: `{"memory_quota":2048,"case_sensitive":true,"filter":{"rules":["*.*"]},"sink":{"protocol":"a","date_separator":"a"}}`, err: false, expectedReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, CaseSensitive: pointer.ToBool(true), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", DateSeparator: "a", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, }, { srcReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, CaseSensitive: pointer.ToBool(false), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(false), }, }, }, replicaConfigStr: `{"memory_quota":2048,"case_sensitive":true,"filter":{"rules":["*.*"]},"sink":{"protocol":"a","date_separator":"a","debezium":{"output_old_value":true}}}`, err: false, expectedReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, CaseSensitive: pointer.ToBool(false), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", DateSeparator: "a", Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(false), }, }, }, }, { srcReplicaConfig: &ticdc_v1.ReplicaConfig{}, replicaConfigStr: `{"memory_quota":2048,"case_sensitive":true,"filter":{"rules":["*.*"]},"sink":{"protocol":"a","date_separator":"a"}}`, err: false, expectedReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 2048, CaseSensitive: pointer.ToBool(true), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"*.*"}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "a", DateSeparator: "a", }, }, }, } for _, tc := range testcases { replicaConfig, err := s.model.overrideChangefeedReplicaConfig(tc.srcReplicaConfig, tc.replicaConfigStr) if tc.err { s.Error(err) } else { s.NoError(err) s.EqualExportedValues(tc.expectedReplicaConfig, replicaConfig) } } } func (s *changefeedModelRegionalImplTestSuite) TestCreateChangefeedByRegional() { var ( changefeedID = "1111" startTS = uint64(123123) uriConfig = &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", } replicaConfig = &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, } ticdcClusterEnt = &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } ) s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() err := s.model.createChangefeedByRegional( context.Background(), changefeedID, startTS, uriConfig, replicaConfig, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "CreateChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.CreateChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", Changefeed: &ticdc_v1.ChangefeedConfig{ ChangefeedId: "1111", StartTs: 123123, SinkUri: "kafka://kafka:9092/test?protocol=&enable-tls=false&replication-factor=0&partition-num=0&kafka-version=", ReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, }, }, }). Return(nil, nil) defer call.Unset() err := s.model.createChangefeedByRegional( context.Background(), changefeedID, startTS, uriConfig, replicaConfig, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestConvertRegionalChangefeedReplicaConfig() { testcases := []struct { replicaConfig *clusterproxy.ReplicaConfig expectedRegionalConfig *ticdc_v1.ReplicaConfig }{ { replicaConfig: &clusterproxy.ReplicaConfig{}, expectedRegionalConfig: &ticdc_v1.ReplicaConfig{}, }, { replicaConfig: &clusterproxy.ReplicaConfig{ MemoryQuota: 1024, EnableOldValue: pointer.ToBool(true), }, expectedRegionalConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, EnableOldValue: pointer.ToBool(true), }, }, { replicaConfig: &clusterproxy.ReplicaConfig{ MemoryQuota: 1024, EnableOldValue: pointer.ToBool(true), Filter: &clusterproxy.FilterConfig{ Rules: []string{"test.*"}, EventFilters: []clusterproxy.EventFilterRule{ { Matcher: []string{"test.*"}, }, }, }, Mounter: &clusterproxy.MounterConfig{ WorkerNum: 16, }, Sink: &clusterproxy.SinkConfigV2{ Protocol: "abc", CSVConfig: &clusterproxy.CSVConfig{ Delimiter: "a", OutputFieldHeader: pointer.ToBool(true), }, DispatchRules: []*clusterproxy.DispatchRuleV2{ { Matcher: []string{"test.*"}, Topic: "test", }, }, ColumnSelectors: []*clusterproxy.ColumnSelector{ { Matcher: []string{"test.a"}, Columns: []string{"b"}, }, }, KafkaSink: &clusterproxy.KafkaConfig{ OutputRawChangeEvent: pointer.ToBool(false), }, CloudStorageConfig: &clusterproxy.CloudStorageConfig{ OutputRawChangeEvent: pointer.ToBool(true), }, }, Consistent: &clusterproxy.ConsistentConfig{ Level: "a", MaxLogSize: 1024, }, }, expectedRegionalConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, EnableOldValue: pointer.ToBool(true), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{ { Matcher: []string{"test.*"}, }, }, }, Mounter: &ticdc_v1.ReplicaConfig_MounterConfig{ WorkerNum: 16, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Csv: &ticdc_v1.ReplicaConfig_CSVConfig{ Delimiter: "a", OutputFieldHeader: pointer.ToBool(true), }, Dispatchers: []*ticdc_v1.ReplicaConfig_DispatchRule{ { Matcher: []string{"test.*"}, Topic: "test", }, }, ColumnSelectors: []*ticdc_v1.ReplicaConfig_ColumnSelector{ { Matcher: []string{"test.a"}, Columns: []string{"b"}, }, }, KafkaSink: &ticdc_v1.ReplicaConfig_KafkaConfig{ OutputRawChangeEvent: pointer.ToBool(false), }, CloudStorageConfig: &ticdc_v1.ReplicaConfig_CloudStorageConfig{ OutputRawChangeEvent: pointer.ToBool(true), }, }, Consistent: &ticdc_v1.ReplicaConfig_ConsistentConfig{ Level: "a", MaxLogSize: 1024, }, }, }, { replicaConfig: &clusterproxy.ReplicaConfig{ ForceReplicate: pointer.ToBool(true), Filter: &clusterproxy.FilterConfig{ Rules: []string{"*.*"}, EventFilters: []clusterproxy.EventFilterRule{ { Matcher: []string{"*.*"}, IgnoreEvent: []string{"delete"}, }, }, }, Sink: &clusterproxy.SinkConfigV2{ DispatchRules: []*clusterproxy.DispatchRuleV2{ { Matcher: []string{"*.*"}, Topic: "", Partition: "ts", }, }, PulsarConfig: &clusterproxy.PulsarConfig{ AuthenticationToken: "", BasicUserName: "user", BasicPassword: "password", CompressionType: "lz4", ConnectionTimeout: 5, OperationTimeout: 4, SendTimeout: 3, BatchingMaxMessages: 100, BatchingMaxPublishDelay: 1000, Oauth2: &clusterproxy.PulsarOAuth2{}, OutputRawChangeEvent: pointer.ToBool(true), }, }, }, expectedRegionalConfig: &ticdc_v1.ReplicaConfig{ ForceReplicate: pointer.ToBool(true), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"*.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{ { Matcher: []string{"*.*"}, IgnoreEvent: []string{"delete"}, }, }, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Dispatchers: []*ticdc_v1.ReplicaConfig_DispatchRule{ { Matcher: []string{"*.*"}, Topic: "", Partition: "ts", }, }, ColumnSelectors: []*ticdc_v1.ReplicaConfig_ColumnSelector{}, PulsarConfig: &ticdc_v1.ReplicaConfig_PulsarConfig{ AuthenticationToken: "", BasicUserName: "user", BasicPassword: "password", CompressionType: "lz4", ConnectionTimeout: 5, OperationTimeout: 4, SendTimeout: 3, BatchingMaxMessages: 100, BatchingMaxPublishDelay: 1000, OutputRawChangeEvent: pointer.ToBool(true), }, }, }, }, } for _, tc := range testcases { regionalConfig := s.model.normalizeChangefeedReplicaConfig(tc.replicaConfig) s.Equal(tc.expectedRegionalConfig, regionalConfig) } } func (s *changefeedModelRegionalImplTestSuite) TestAssembleChangefeedEntity() { req := &domain.CreateChangefeedReq{ TiCDCClusterID: 111, KernelChangefeedID: "1111", DownStreamType: entity.TiCDCDownStreamTypeKafka, Config: &domain.TiCDCChangefeedConfig{}, BizType: domain.TiCDCClusterBizTypeChangefeed, BusinessID: 1111, } ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, ProjectID: 11, ClusterID: 111, Name: "cdc-111", Region: "us-west-2", } ticdcClusterEnt.ID = 111 expectedChangefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", TenantID: 1, ProjectID: 11, UpStreamClusterID: 111, TiCDCClusterID: 111, DownStreamType: entity.TiCDCDownStreamTypeKafka, BusinessType: entity.TiCDCClusterBizTypeChangefeed, BusinessID: 1111, } changefeedEnt := s.model.assembleChangefeedEntity(req, ticdcClusterEnt) s.Equal(expectedChangefeedEnt, changefeedEnt) } func (s *changefeedModelRegionalImplTestSuite) TestAssembleKernelChangefeedConfig() { changefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", DownStreamType: entity.TiCDCDownStreamTypeKafka, } changefeedEnt.ID = 1111 changefeedConfig := &domain.TiCDCChangefeedConfig{ URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", Compression: pointer.ToString("gzip"), }, Config: &domain.TiCDCChangefeedConfigData{ StartTs: pointer.ToUint64(123123), ReplicaConfig: &clusterproxy.ReplicaConfig{ MemoryQuota: 1024, Filter: &clusterproxy.FilterConfig{ Rules: []string{"test.*"}, EventFilters: []clusterproxy.EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &clusterproxy.SinkConfigV2{ Protocol: "abc", DebeziumConfig: &clusterproxy.DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, }, } s.Run("get changefeed config error", func() { mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error") call := s.mockConfigManager.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, _, err := s.model.assembleKernelChangefeedConfig(context.Background(), changefeedEnt, changefeedConfig) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "get changefeed config") }) getConfCall := s.mockConfigManager.On("Get", mock.Anything, &domain.GetChangefeedConfigReq{ ChangefeedID: 1111, }).Return(&domain.GetChangefeedConfigResp{ Configuration: `{}`, URIConfiguration: `{}`, }, nil) defer getConfCall.Unset() expectedReplicaConfig := &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Protocol: "abc", Dispatchers: []*ticdc_v1.ReplicaConfig_DispatchRule{}, ColumnSelectors: []*ticdc_v1.ReplicaConfig_ColumnSelector{}, Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, } s.Run("assemble kernel changefeed config successfully", func() { uriConfig, replicaConfig, err := s.model.assembleKernelChangefeedConfig( context.Background(), changefeedEnt, changefeedConfig) s.NoError(err) s.Equal(changefeedConfig.URIConfig, uriConfig) s.EqualExportedValues(expectedReplicaConfig, replicaConfig) }) } // TestPersistCreateIntent covers the Phase-1 helper of the new 3-phase create // flow: load existing pending row by deterministic kernel ID, otherwise insert // a fresh row + config in a short transaction. func (s *changefeedModelRegionalImplTestSuite) TestPersistCreateIntent() { ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()). DoAndReturn(func(ctx context.Context, fn func(context.Context) error) error { return fn(ctx) }).AnyTimes() ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, ProjectID: 11, ClusterID: 111, Name: "cdc-111", Region: "us-west-2", Version: "v7.5.0", } ticdcClusterEnt.ID = 111 req := &domain.CreateChangefeedReq{ TiCDCClusterID: 111, KernelChangefeedID: "1111", DownStreamType: entity.TiCDCDownStreamTypeKafka, BizType: domain.TiCDCClusterBizTypeChangefeed, BusinessID: 1111, Config: &domain.TiCDCChangefeedConfig{}, } s.Run("returns existing row on retry without inserting", func() { existing := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", KernelStatus: entity.TiCDCChangefeedKernelStatusPending, } existing.ID = 1111 getCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(existing, nil) defer getCall.Unset() ent, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.NoError(err) s.Same(existing, ent) }) s.Run("returns existing active row (already converged)", func() { existing := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", KernelStatus: entity.TiCDCChangefeedKernelStatusActive, } existing.ID = 1111 getCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(existing, nil) defer getCall.Unset() ent, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.NoError(err) s.Equal(entity.TiCDCChangefeedKernelStatusActive, ent.KernelStatus) }) s.Run("Get returns non-NotFound error is propagated", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "db down") getCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer getCall.Unset() _, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.Error(err) s.Contains(err.Error(), "look up existing changefeed entity") }) notFoundErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") getNotFoundCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(nil, notFoundErr) defer getNotFoundCall.Unset() s.Run("save error inside intent tx propagates", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "save fail") saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr) defer saveCall.Unset() _, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.Error(err) s.Contains(err.Error(), "save changefeed entity") }) saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil). Run(func(args mock.Arguments) { ent := args[1].(*entity.TiCDCChangefeed) ent.ID = 1111 }) defer saveCall.Unset() s.Run("config create error inside intent tx propagates", func() { mockErr := bizerr.New(bizerr.ErrCodeGRPC, "config-center down") cfgCall := s.mockConfigManager.On("Create", mock.Anything, mock.Anything).Return(nil, mockErr) defer cfgCall.Unset() _, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.Error(err) s.Contains(err.Error(), "create changefeed config") }) cfgCall := s.mockConfigManager.On("Create", mock.Anything, &domain.CreateChangefeedConfigReq{ ChangefeedID: 1111, DownstreamType: entity.TiCDCDownStreamTypeKafka, }).Return(nil, nil) defer cfgCall.Unset() s.Run("first call inserts a pending row", func() { ent, err := s.model.persistCreateIntent(context.Background(), req, ticdcClusterEnt) s.NoError(err) s.Equal(entity.TiCDCChangefeedKernelStatusPending, ent.KernelStatus, "new row must be persisted as pending — kernel call comes after this commit") }) } // TestCreateOrVerifyKernelChangefeed covers the Phase-2 helper: regional create // + idempotency probe. Mirrors the cases that used to live in TestDoCreateChangefeed. func (s *changefeedModelRegionalImplTestSuite) TestCreateOrVerifyKernelChangefeed() { ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, ProjectID: 11, ClusterID: 111, Name: "cdc-111", Region: "us-west-2", Version: "v7.5.0", } ticdcClusterEnt.ID = 111 changefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", TenantID: 1, ProjectID: 11, UpStreamClusterID: 111, TiCDCClusterID: 111, DownStreamType: entity.TiCDCDownStreamTypeKafka, BusinessType: entity.TiCDCClusterBizTypeChangefeed, BusinessID: 1111, KernelStatus: entity.TiCDCChangefeedKernelStatusPending, } changefeedEnt.ID = 1111 changefeedConfig := &domain.TiCDCChangefeedConfig{ URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", Compression: pointer.ToString("gzip"), }, Config: &domain.TiCDCChangefeedConfigData{ StartTs: pointer.ToUint64(123123), ReplicaConfig: &clusterproxy.ReplicaConfig{ MemoryQuota: 1024, Filter: &clusterproxy.FilterConfig{ Rules: []string{"test.*"}, EventFilters: []clusterproxy.EventFilterRule{{Matcher: []string{"test.*"}}}, }, Sink: &clusterproxy.SinkConfigV2{ Protocol: "abc", DebeziumConfig: &clusterproxy.DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, }, } s.Run("get changefeed config error", func() { mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error") call := s.mockConfigManager.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "get changefeed config") }) getConfCall := s.mockConfigManager.On("Get", mock.Anything, &domain.GetChangefeedConfigReq{ ChangefeedID: 1111, }).Return(&domain.GetChangefeedConfigResp{ Configuration: `{}`, URIConfiguration: `{}`, }, nil) defer getConfCall.Unset() s.Run("call regional error and kernel list confirms changefeed absent", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() listCall := s.mockChangefeedCli. On("ListChangefeeds", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.ListChangefeedsResponse{ Items: []*ticdc_v1.ChangefeedCommonInfo{{Id: "some-other-cf"}}, }, nil) defer listCall.Unset() err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(appconfig.ErrCodeCreateKernelChangefeedFailed, err.RawErrCode()) s.Equal("network error", err.ErrMsg()) }) s.Run("regional reports error but kernel list shows the changefeed (timeout race)", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "context canceled") call := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() listCall := s.mockChangefeedCli. On("ListChangefeeds", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.ListChangefeedsResponse{ Items: []*ticdc_v1.ChangefeedCommonInfo{ {Id: changefeedEnt.KernelChangefeedID}, }, }, nil) defer listCall.Unset() err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.NoError(err, "kernel orphan must be reconciled into success") }) s.Run("regional reports error and probe also fails returns original error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() probeErr := bizerr.New(bizerr.ErrCodeUnknown, "probe also failed") listCall := s.mockChangefeedCli. On("ListChangefeeds", mock.Anything, mock.Anything, mock.Anything). Return(nil, probeErr) defer listCall.Unset() err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(appconfig.ErrCodeCreateKernelChangefeedFailed, err.RawErrCode()) s.Equal("network error", err.ErrMsg()) }) s.Run("changefeed already exists short-circuit skips probe", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, `Failed to create changefeed: {\n \"error_msg\": \"[CDC:ErrChangeFeedAlreadyExists]changefeed already exists, changefeed-1896979993011191808\",\n \"error_code\": \"CDC:ErrChangeFeedAlreadyExists\"\n}`) call := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.NoError(err) }) regionalCall := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, nil) defer regionalCall.Unset() s.Run("create changefeed successfully", func() { err := s.model.createOrVerifyKernelChangefeed(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestCreateChangefeed() { req := &domain.CreateChangefeedReq{ TiCDCClusterID: 111, KernelChangefeedID: "1111", DownStreamType: entity.TiCDCDownStreamTypeKafka, BizType: domain.TiCDCClusterBizTypeChangefeed, BusinessID: 1111, Config: &domain.TiCDCChangefeedConfig{ URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", Compression: pointer.ToString("gzip"), }, Config: &domain.TiCDCChangefeedConfigData{ ReplicaConfig: &clusterproxy.ReplicaConfig{}, }, }, } s.Run("load ticdc cluster entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.CreateChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "query ticdc cluster entity") }) ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, ProjectID: 11, ClusterID: 111, Name: "cdc-111", Region: "us-west-2", } ticdcClusterEnt.ID = 111 getCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, uint64(111)).Return(ticdcClusterEnt, nil) defer getCall.Unset() s.Run("intent transaction error", func() { notFoundErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") getCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(nil, notFoundErr) defer getCall.Unset() ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr) _, err := s.model.CreateChangefeed(context.Background(), req) s.Error(err) }) s.Run("active row short-circuits without re-issuing kernel call", func() { existing := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", KernelStatus: entity.TiCDCChangefeedKernelStatusActive, } existing.ID = 1111 getCall := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(existing, nil) defer getCall.Unset() // No CreateChangefeed / Save mocks set up — strict mock would fail if called. resp, err := s.model.CreateChangefeed(context.Background(), req) s.NoError(err) s.Equal(uint64(1111), resp.ChangefeedID) }) notFoundErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") getNotFound := s.mockChangefeedRepo.On("Get", mock.Anything, mock.Anything, mock.Anything). Return(nil, notFoundErr) defer getNotFound.Unset() var savedStatuses []entity.TiCDCChangefeedKernelStatus saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil). Run(func(args mock.Arguments) { ent := args[1].(*entity.TiCDCChangefeed) if ent.ID == 0 { ent.ID = 1111 } savedStatuses = append(savedStatuses, ent.KernelStatus) }) defer saveCall.Unset() cfgCall := s.mockConfigManager.On("Create", mock.Anything, mock.Anything).Return(nil, nil) defer cfgCall.Unset() getConfCall := s.mockConfigManager.On("Get", mock.Anything, mock.Anything). Return(&domain.GetChangefeedConfigResp{Configuration: `{}`, URIConfiguration: `{}`}, nil) defer getConfCall.Unset() ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) // DoAndReturn invokes the inner closure so Save / configManager.Create // inside Phase 1 actually fires through to the mocks below — without this // shortcut Save would never be observed. mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()). DoAndReturn(func(ctx context.Context, fn func(context.Context) error) error { return fn(ctx) }).AnyTimes() s.Run("kernel error keeps row pending and surfaces error for DAG retry", func() { savedStatuses = nil // Phase 2 fails; row must NOT be flipped to active. DAG will retry. mockErr := bizerr.New(bizerr.ErrCodeUnknown, "regional down") regionalCall := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer regionalCall.Unset() listCall := s.mockChangefeedCli. On("ListChangefeeds", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.ListChangefeedsResponse{}, nil) defer listCall.Unset() _, err := s.model.CreateChangefeed(context.Background(), req) s.Error(err) s.Equal(appconfig.ErrCodeCreateKernelChangefeedFailed, err.RawErrCode()) // Only the Phase-1 save (status=pending) happened; no Phase-3 flip. s.Equal([]entity.TiCDCChangefeedKernelStatus{entity.TiCDCChangefeedKernelStatusPending}, savedStatuses, "row must stay pending when kernel call fails — DAG retry will pick it up") }) regionalCall := s.mockChangefeedCli. On("CreateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, nil) defer regionalCall.Unset() s.Run("create changefeed successfully flips row to active", func() { savedStatuses = nil _, err := s.model.CreateChangefeed(context.Background(), req) s.NoError(err) s.Equal([]entity.TiCDCChangefeedKernelStatus{ entity.TiCDCChangefeedKernelStatusPending, // Phase 1 save entity.TiCDCChangefeedKernelStatusActive, // Phase 3 save }, savedStatuses) }) } func (s *changefeedModelRegionalImplTestSuite) TestDeleteChangefeedByRegional() { changefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", } ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("DeleteChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.deleteChangefeedByRegional(context.Background(), changefeedEnt, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "DeleteChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.DeleteChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", }). Return(nil) defer call.Unset() err := s.model.deleteChangefeedByRegional(context.Background(), changefeedEnt, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestDoDeleteChangefeed() { changefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", BusinessID: entity.ID(1234), } ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", } s.Run("save changefeed error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr) defer call.Unset() err := s.model.doDeleteChangefeed(context.Background(), changefeedEnt, ticdcClusterEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "save changefeed entity") }) saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil) defer saveCall.Unset() s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("DeleteChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.doDeleteChangefeed(context.Background(), changefeedEnt, ticdcClusterEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "delete changefeed by regional") }) regionalCall := s.mockChangefeedCli. On("DeleteChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil) defer regionalCall.Unset() s.Run("deleteChangefeedSuccessfully", func() { err := s.model.doDeleteChangefeed(context.Background(), changefeedEnt, ticdcClusterEnt) s.NoError(err) s.NotNil(changefeedEnt.DeletedAt) }) } func (s *changefeedModelRegionalImplTestSuite) TestDeleteChangefeed() { req := &domain.DeleteChangefeedReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.DeleteChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) s.Run("changefeed entity not found", func() { mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.DeleteChangefeed(context.Background(), req) s.NoError(err) }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } changefeedEnt.ID = 1111 s.Run("changefeed entity is deleted", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() _, err := s.model.DeleteChangefeed(context.Background(), req) s.NoError(err) }) changefeedEnt.DeletedAt = nil loadCall := s.mockChangefeedRepo.On("Load", mock.Anything, uint64(1111)).Return(changefeedEnt, nil) defer loadCall.Unset() s.Run("load ticdc cluster entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.DeleteChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load ticdc cluster entity") }) ticdcClusterEnt := &entity.TiCDCCluster{} cdcCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer cdcCall.Unset() s.Run("transcation error", func() { ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr) _, err := s.model.DeleteChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr, err) }) ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil) s.Run("delete changefeed successfully", func() { _, err := s.model.DeleteChangefeed(context.Background(), req) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestModifyChangefeedConfigByRegional() { var ( changefeedID = "1111" uriConfig = &domain.TiCDCChangefeedKafkaURIConfigData{ BrokerEndpoints: "kafka:9092", DefaultTopic: "test", } replicaConfig = &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, } ticdcClusterEnt = &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } ) s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("UpdateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.modifyChangefeedConfigByRegional( context.Background(), changefeedID, uriConfig, replicaConfig, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "UpdateChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.UpdateChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", Changefeed: &ticdc_v1.ChangefeedConfig{ SinkUri: "kafka://kafka:9092/test?protocol=&enable-tls=false&replication-factor=0&partition-num=0&kafka-version=", ReplicaConfig: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, }, }, }). Return(nil) defer call.Unset() err := s.model.modifyChangefeedConfigByRegional( context.Background(), changefeedID, uriConfig, replicaConfig, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestGetChangefeedFromRegional() { var ( changefeedID = "1111" ticdcClusterEnt = &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } ) s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() _, err := s.model.getChangefeedFromRegional(context.Background(), changefeedID, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "GetChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.GetChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", }). Return(&ticdc_v1.Changefeed{}, nil) defer call.Unset() _, err := s.model.getChangefeedFromRegional(context.Background(), changefeedID, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestDoModifyChangefeedConfig() { changefeedConfig := &domain.TiCDCChangefeedConfig{ URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{}, Config: &domain.TiCDCChangefeedConfigData{ ReplicaConfig: &clusterproxy.ReplicaConfig{ MemoryQuota: 1024, CaseSensitive: pointer.ToBool(false), Filter: &clusterproxy.FilterConfig{ Rules: []string{"test.*"}, }, Sink: &clusterproxy.SinkConfigV2{ DebeziumConfig: &clusterproxy.DebeziumConfig{ OutputOldValue: pointer.ToBool(false), }, }, }, }, } changefeedEnt := &entity.TiCDCChangefeed{ DownStreamType: entity.TiCDCDownStreamTypeKafka, KernelChangefeedID: "1111", } changefeedEnt.ID = 1111 ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", } s.Run("get kernel changefeed error", func() { mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error") call := s.mockChangefeedCli. On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() err := s.model.doModifyChangefeedConfig(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "get kernel changefeed") }) mockKernelChangefeed := &ticdc_v1.Changefeed{ Config: &ticdc_v1.ReplicaConfig{ MemoryQuota: 512, CaseSensitive: pointer.ToBool(true), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"*.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{ { Matcher: []string{"test.*"}, IgnoreEvent: []string{"delete"}, }, }, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ EncoderConcurrency: 32, Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(true), }, }, }, } getChangefeedCall := s.mockChangefeedCli. On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockKernelChangefeed, nil) defer getChangefeedCall.Unset() s.Run("get changefeed config error", func() { mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error") call := s.mockConfigManager.On("Get", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() err := s.model.doModifyChangefeedConfig(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "get changefeed config") }) getConfCall := s.mockConfigManager.On("Get", mock.Anything, &domain.GetChangefeedConfigReq{ ChangefeedID: 1111, }).Return(&domain.GetChangefeedConfigResp{ Configuration: `{}`, URIConfiguration: `{}`, }, nil) defer getConfCall.Unset() s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("UpdateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.doModifyChangefeedConfig(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "modify changefeed config by regional") }) expectedReplicaConfig := &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, CaseSensitive: pointer.ToBool(false), Filter: &ticdc_v1.ReplicaConfig_FilterConfig{ Rules: []string{"test.*"}, EventFilters: []*ticdc_v1.ReplicaConfig_EventFilterRule{}, }, Sink: &ticdc_v1.ReplicaConfig_SinkConfig{ Dispatchers: []*ticdc_v1.ReplicaConfig_DispatchRule{}, ColumnSelectors: []*ticdc_v1.ReplicaConfig_ColumnSelector{}, EncoderConcurrency: 32, Debezium: &ticdc_v1.ReplicaConfig_DebeziumConfig{ OutputOldValue: pointer.ToBool(false), }, }, } expectedUpdateReq := &ticdc_v1.UpdateChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", Changefeed: &ticdc_v1.ChangefeedConfig{ SinkUri: changefeedConfig.URIConfig.GenerateURI(), ReplicaConfig: expectedReplicaConfig, }, } regionalCall := s.mockChangefeedCli. On("UpdateChangefeed", mock.Anything, mock.Anything, expectedUpdateReq). Return(nil) defer regionalCall.Unset() s.Run("modify changefeed config successfully", func() { err := s.model.doModifyChangefeedConfig(context.Background(), changefeedConfig, ticdcClusterEnt, changefeedEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestModifyChangefeedConfig() { req := &domain.ModifyChangefeedConfigReq{ ChangefeedID: 1111, Config: &domain.TiCDCChangefeedConfig{ URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{}, Config: &domain.TiCDCChangefeedConfigData{ ReplicaConfig: &clusterproxy.ReplicaConfig{}, }, }, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.ModifyChangefeedConfig(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } changefeedEnt.ID = 1111 loadCall := s.mockChangefeedRepo.On("Load", mock.Anything, uint64(1111)).Return(changefeedEnt, nil) defer loadCall.Unset() s.Run("changefeed status is not stopped", func() { _, err := s.model.ModifyChangefeedConfig(context.Background(), req) s.Error(err) s.Contains(err.Error(), "is not allowed to update") }) changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusFailed s.Run("load ticdc cluster entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.ModifyChangefeedConfig(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load ticdc cluster entity") }) ticdcClusterEnt := &entity.TiCDCCluster{} cdcCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer cdcCall.Unset() getConfCall := s.mockConfigManager.On("Get", mock.Anything, mock.Anything). Return(&domain.GetChangefeedConfigResp{ Configuration: `{}`, URIConfiguration: `{}`, }, nil) defer getConfCall.Unset() getChangefeedCall := s.mockChangefeedCli. On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.Changefeed{Config: &ticdc_v1.ReplicaConfig{}}, nil) defer getChangefeedCall.Unset() s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("UpdateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() _, err := s.model.ModifyChangefeedConfig(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "modify changefeed config") }) regionalCall := s.mockChangefeedCli. On("UpdateChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil) defer regionalCall.Unset() s.Run("modify changefeed config successfully", func() { _, err := s.model.ModifyChangefeedConfig(context.Background(), req) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestPauseChangefeedByRegional() { changefeedID := "1111" ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("PauseChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.pauseChangefeedByRegional(context.Background(), changefeedID, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "PauseChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.PauseChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", }). Return(nil) defer call.Unset() err := s.model.pauseChangefeedByRegional(context.Background(), changefeedID, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestDoPauseChangefeed() { mockChangefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } mockTiCDCClusterEnt := &entity.TiCDCCluster{} s.Run("save changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr) defer call.Unset() err := s.model.doPauseChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.Error(err) s.Contains(err.Error(), "update changefeed entity") }) expectedChangefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, KernelMetrics: entity.TiCDCChangefeedKernelMetrics{ FeedState: entity.ChangefeedKernelStatusStopped, }, } saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, expectedChangefeedEnt).Return(nil) defer saveCall.Unset() s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("PauseChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.doPauseChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "pause changefeed by regional") }) regionalCall := s.mockChangefeedCli. On("PauseChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil) defer regionalCall.Unset() s.Run("pause changefeed successfully", func() { err := s.model.doPauseChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestPauseChangefeed() { req := &domain.PauseChangefeedReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.PauseChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } changefeedEnt.ID = 1111 loadCall := s.mockChangefeedRepo.On("Load", mock.Anything, uint64(1111)).Return(changefeedEnt, nil) defer loadCall.Unset() s.Run("changefeed is already stopped", func() { changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped _, err := s.model.PauseChangefeed(context.Background(), req) s.NoError(err) }) changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusFailed s.Run("load ticdc cluster entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.PauseChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load ticdc cluster entity") }) ticdcClusterEnt := &entity.TiCDCCluster{} cdcCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer cdcCall.Unset() s.Run("transcation error", func() { ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr) _, err := s.model.PauseChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr, err) }) ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()). DoAndReturn(func(ctx context.Context, fn func(context.Context) error) error { return fn(ctx) }) changefeedEnt.KernelChangefeedID = "cf-1111" ticdcClusterEnt.TenantID = 1 ticdcClusterEnt.Name = "cdc-111" ticdcClusterEnt.Region = "us-west-2" ticdcClusterEnt.InfraProviderType = "eks" saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, changefeedEnt).Return(nil).Once() defer saveCall.Unset() pauseCall := s.mockChangefeedCli.On( "PauseChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.PauseChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "cf-1111", }, ).Return(nil).Once() defer pauseCall.Unset() s.Run("pause changefeed successfully", func() { _, err := s.model.PauseChangefeed(context.Background(), req) s.NoError(err) s.Equal(entity.ChangefeedKernelStatusStopped, changefeedEnt.KernelMetrics.FeedState) }) } func (s *changefeedModelRegionalImplTestSuite) TestResumeChangefeedByRegional() { changefeedID := "1111" ticdcClusterEnt := &entity.TiCDCCluster{ TenantID: 1, Name: "cdc-111", Region: "us-west-2", InfraProviderType: "eks", } s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("ResumeChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.resumeChangefeedByRegional(context.Background(), changefeedID, ticdcClusterEnt) s.Error(err) s.Equal(mockErr, err) }) s.Run("call regional successfully", func() { call := s.mockChangefeedCli.On( "ResumeChangefeed", mock.Anything, ®ionalapi.Region{ TenantID: 1, Name: "us-west-2", CloudProvider: "aws", }, &ticdc_v1.ResumeChangefeedRequest{ TicdcClusterId: "cdc-111", ChangefeedId: "1111", Config: &ticdc_v1.ResumeChangefeedConfig{}, }). Return(nil) defer call.Unset() err := s.model.resumeChangefeedByRegional(context.Background(), changefeedID, ticdcClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestDoResumeChangefeed() { mockChangefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } mockTiCDCClusterEnt := &entity.TiCDCCluster{} s.Run("save changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr) defer call.Unset() err := s.model.doResumeChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.Error(err) s.Contains(err.Error(), "update changefeed entity") }) expectedChangefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, KernelMetrics: entity.TiCDCChangefeedKernelMetrics{ FeedState: entity.ChangefeedKernelStatusNormal, }, } saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, expectedChangefeedEnt).Return(nil) defer saveCall.Unset() s.Run("call regional error", func() { mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error") call := s.mockChangefeedCli. On("ResumeChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(mockErr) defer call.Unset() err := s.model.doResumeChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "resume changefeed by regional") }) regionalCall := s.mockChangefeedCli. On("ResumeChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(nil) defer regionalCall.Unset() s.Run("resume changefeed successfully", func() { err := s.model.doResumeChangefeed(context.Background(), mockChangefeedEnt, mockTiCDCClusterEnt) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestResumeChangefeed() { req := &domain.ResumeChangefeedReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.ResumeChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } changefeedEnt.ID = 1111 loadCall := s.mockChangefeedRepo.On("Load", mock.Anything, uint64(1111)).Return(changefeedEnt, nil) defer loadCall.Unset() s.Run("changefeed is already normal", func() { changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusNormal _, err := s.model.ResumeChangefeed(context.Background(), req) s.NoError(err) }) changefeedEnt.KernelMetrics.FeedState = entity.ChangefeedKernelStatusStopped s.Run("load ticdc cluster entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.ResumeChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load ticdc cluster entity") }) ticdcClusterEnt := &entity.TiCDCCluster{} cdcCall := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer cdcCall.Unset() s.Run("transcation error", func() { ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(mockErr) _, err := s.model.ResumeChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr, err) }) ctrl := gomock.NewController(s.T()) defer s.cleanTransProvider() mockTransProvider := s.newTransProvider(ctrl) mockTransProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).Return(nil) s.Run("resume changefeed successfully", func() { _, err := s.model.ResumeChangefeed(context.Background(), req) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestListChangefeeds() { req := &domain.ListChangefeedsReq{} s.Run("list changefeed entities error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo. On("List", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() _, err := s.model.ListChangefeeds(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "list changefeed entities") }) changefeedEnts := []entity.TiCDCChangefeed{ { Updatable: entity.Updatable{ Base: entity.Base{ ID: 1111, }, }, KernelChangefeedID: "1111", }, { Updatable: entity.Updatable{ Base: entity.Base{ ID: 1112, }, }, KernelChangefeedID: "1112", }, } listCall := s.mockChangefeedRepo. On("List", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(changefeedEnts, nil) defer listCall.Unset() s.Run("count changefeed entities error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Count", mock.Anything, mock.Anything).Return(int64(0), mockErr) defer call.Unset() _, err := s.model.ListChangefeeds(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "count changefeed entities") }) count := int64(2) countCall := s.mockChangefeedRepo.On("Count", mock.Anything, mock.Anything).Return(count, nil) defer countCall.Unset() s.Run("list changefeeds successfully", func() { resp, err := s.model.ListChangefeeds(context.Background(), req) s.NoError(err) s.Equal(count, resp.Total) s.Equal(len(changefeedEnts), len(resp.Items)) for i, item := range resp.Items { s.Equal(changefeedEnts[i], item.ChangefeedEntity) } }) } func (s *changefeedModelRegionalImplTestSuite) TestQueryChangefeed() { req := &domain.QueryChangefeedReq{} s.Run("get changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo. On("Get", mock.Anything, mock.Anything, mock.Anything). Return(nil, mockErr) defer call.Unset() _, err := s.model.QueryChangefeed(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "query changefeed entity") }) changefeedEnt := &entity.TiCDCChangefeed{ KernelChangefeedID: "1111", } changefeedEnt.ID = 1111 call := s.mockChangefeedRepo. On("Get", mock.Anything, mock.Anything, mock.Anything). Return(changefeedEnt, nil) defer call.Unset() s.Run("query changefeed successfully", func() { resp, err := s.model.QueryChangefeed(context.Background(), req) s.NoError(err) s.Equal(*changefeedEnt, resp.ChangefeedEntity) }) } func (s *changefeedModelRegionalImplTestSuite) TestMarkChangefeedDeleted() { req := &domain.MarkChangefeedDeletedReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.MarkChangefeedDeleted(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) s.Run("changefeed entity not found", func() { mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.MarkChangefeedDeleted(context.Background(), req) s.NoError(err) }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, } changefeedEnt.ID = 1111 s.Run("changefeed entity is deleted", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() _, err := s.model.MarkChangefeedDeleted(context.Background(), req) s.NoError(err) }) changefeedEnt.DeletedAt = nil loadCall := s.mockChangefeedRepo.On("Load", mock.Anything, uint64(1111)).Return(changefeedEnt, nil) defer loadCall.Unset() s.Run("save changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(mockErr) defer call.Unset() _, err := s.model.MarkChangefeedDeleted(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "save changefeed entity") }) saveCall := s.mockChangefeedRepo.On("Save", mock.Anything, mock.Anything).Return(nil) defer saveCall.Unset() s.Run("mark changefeed deleted successfully", func() { _, err := s.model.MarkChangefeedDeleted(context.Background(), req) s.NoError(err) s.NotNil(changefeedEnt.DeletedAt) }) } func (s *changefeedModelRegionalImplTestSuite) TestGetChangefeedDetail() { req := &domain.GetChangefeedDetailReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.GetChangefeedDetail(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) s.Run("changefeed entity not found", func() { mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.GetChangefeedDetail(context.Background(), req) s.Error(err) }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, KernelChangefeedID: "cf-111", } changefeedEnt.ID = 1111 s.Run("load cluster failed", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call2.Unset() _, err := s.model.GetChangefeedDetail(context.Background(), req) s.Error(err) }) ticdcClusterEnt := &entity.TiCDCCluster{ Name: "cdc-111", } s.Run("call regional failed", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer call2.Unset() mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call3 := s.mockChangefeedCli.On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything).Return(nil, mockErr) defer call3.Unset() _, err := s.model.GetChangefeedDetail(context.Background(), req) s.Error(err) }) s.Run("successfully", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer call2.Unset() call3 := s.mockChangefeedCli.On("GetChangefeed", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.Changefeed{Id: "cf-id"}, nil) defer call3.Unset() _, err := s.model.GetChangefeedDetail(context.Background(), req) s.NoError(err) }) } func (s *changefeedModelRegionalImplTestSuite) TestGetChangefeedCluster() { req := &domain.GetChangefeedClusterReq{ ChangefeedID: 1111, } s.Run("load changefeed entity error", func() { mockErr := bizerr.New(bizerr.ErrCodeDB, "network error") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.GetTiCDCCluster(context.Background(), req) s.Error(err) s.Equal(mockErr.RawErrCode(), err.RawErrCode()) s.Contains(err.Error(), "load changefeed entity") }) s.Run("changefeed entity not found", func() { mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call.Unset() _, err := s.model.GetTiCDCCluster(context.Background(), req) s.Error(err) }) changefeedEnt := &entity.TiCDCChangefeed{ TiCDCClusterID: 111, KernelChangefeedID: "cf-111", } changefeedEnt.ID = 1111 s.Run("load cluster failed", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(nil, mockErr) defer call2.Unset() _, err := s.model.GetTiCDCCluster(context.Background(), req) s.Error(err) }) ticdcClusterEnt := &entity.TiCDCCluster{ Name: "cdc-111", } s.Run("call regional failed", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer call2.Unset() mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found") call3 := s.mockTiCDCClusterCli.On("GetTiCDCCluster", mock.Anything, mock.Anything, mock.Anything).Return(nil, mockErr) defer call3.Unset() _, err := s.model.GetTiCDCCluster(context.Background(), req) s.Error(err) }) s.Run("successfully", func() { deletedAt := time.Now() changefeedEnt.DeletedAt = &deletedAt call := s.mockChangefeedRepo.On("Load", mock.Anything, mock.Anything).Return(changefeedEnt, nil) defer call.Unset() call2 := s.mockTiCDCClusterRepo.On("Load", mock.Anything, mock.Anything).Return(ticdcClusterEnt, nil) defer call2.Unset() call3 := s.mockTiCDCClusterCli.On("GetTiCDCCluster", mock.Anything, mock.Anything, mock.Anything). Return(&ticdc_v1.TiCDCCluster{Id: "cdc-id"}, nil) defer call3.Unset() _, err := s.model.GetTiCDCCluster(context.Background(), req) s.NoError(err) }) } func TestPbToStruct(t *testing.T) { cf := &ticdc_v1.Changefeed{ Id: "1111", Config: &ticdc_v1.ReplicaConfig{ MemoryQuota: 1024, }, } pbStruct, err := pbToStruct(cf) require.Nil(t, err) data, err := protojson.Marshal(pbStruct) require.Nil(t, err) var cf2 = new(ticdc_v1.Changefeed) require.Nil(t, protojson.Unmarshal(data, cf2)) require.Equal(t, cf, cf2) } func (s *changefeedModelRegionalImplTestSuite) newTransProvider(ctrl *gomock.Controller) *mock_db.MockTransProvider { mockTransProvider := mock_db.NewMockTransProvider(ctrl) s.model.transProvider = mockTransProvider return mockTransProvider } func (s *changefeedModelRegionalImplTestSuite) cleanTransProvider() { s.model.transProvider = nil } func TestChangefeedModelRegionalImplTestSuite(t *testing.T) { suite.Run(t, new(changefeedModelRegionalImplTestSuite)) }