From ad0c4e1d8d6c2e153dec352e9ed84403928b7d69 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 15:06:14 +0800 Subject: [PATCH 01/13] remove metrics collector mock --- downstreamadapter/sink/kafka/sink_test.go | 9 +++-- pkg/sink/kafka/metrics_collector_mock.go | 47 ----------------------- scripts/generate-mock.sh | 1 - 3 files changed, 5 insertions(+), 52 deletions(-) delete mode 100644 pkg/sink/kafka/metrics_collector_mock.go diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 8b205c1649..b43eb5a38d 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -41,6 +41,10 @@ import ( const kafkaSinkTestTopic = "mock_topic" +type noopMetricsCollector struct{} + +func (noopMetricsCollector) Run(context.Context) {} + func TestSinkWorkersReturnContextError(t *testing.T) { contexts := []struct { name string @@ -168,13 +172,10 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, }, nil) adminClient.EXPECT().Close().AnyTimes() - metricsCollector := kafka.NewMockMetricsCollector(ctrl) - metricsCollector.EXPECT().Run(gomock.Any()).AnyTimes() - factory := kafka.NewMockFactory(ctrl) factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) - factory.EXPECT().MetricsCollector(adminClient).Return(metricsCollector) + factory.EXPECT().MetricsCollector(adminClient).Return(noopMetricsCollector{}) eventRouter, err := eventrouter.NewEventRouter(sinkConfig, topic, false, false) if err != nil { diff --git a/pkg/sink/kafka/metrics_collector_mock.go b/pkg/sink/kafka/metrics_collector_mock.go deleted file mode 100644 index 9e5bc51448..0000000000 --- a/pkg/sink/kafka/metrics_collector_mock.go +++ /dev/null @@ -1,47 +0,0 @@ -// Code generated by MockGen. DO NOT EDIT. -// Source: pkg/sink/kafka/metrics_collector.go - -// Package kafka is a generated GoMock package. -package kafka - -import ( - context "context" - reflect "reflect" - - gomock "github.com/golang/mock/gomock" -) - -// MockMetricsCollector is a mock of MetricsCollector interface. -type MockMetricsCollector struct { - ctrl *gomock.Controller - recorder *MockMetricsCollectorMockRecorder -} - -// MockMetricsCollectorMockRecorder is the mock recorder for MockMetricsCollector. -type MockMetricsCollectorMockRecorder struct { - mock *MockMetricsCollector -} - -// NewMockMetricsCollector creates a new mock instance. -func NewMockMetricsCollector(ctrl *gomock.Controller) *MockMetricsCollector { - mock := &MockMetricsCollector{ctrl: ctrl} - mock.recorder = &MockMetricsCollectorMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockMetricsCollector) EXPECT() *MockMetricsCollectorMockRecorder { - return m.recorder -} - -// Run mocks base method. -func (m *MockMetricsCollector) Run(ctx context.Context) { - m.ctrl.T.Helper() - m.ctrl.Call(m, "Run", ctx) -} - -// Run indicates an expected call of Run. -func (mr *MockMetricsCollectorMockRecorder) Run(ctx interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockMetricsCollector)(nil).Run), ctx) -} diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index c825e85946..28be7895fe 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -36,7 +36,6 @@ fi "$MOCKGEN" -source pkg/sink/codec/simple/marshaller.go -destination pkg/sink/codec/simple/mock/marshaller.go "$MOCKGEN" -source pkg/sink/kafka/cluster_admin_client.go -destination pkg/sink/kafka/cluster_admin_client_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka -"$MOCKGEN" -source pkg/sink/kafka/metrics_collector.go -destination pkg/sink/kafka/metrics_collector_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka "$MOCKGEN" -source pkg/keyspace/keyspace_manager.go -destination pkg/keyspace/keyspace_manager_mock.go -package keyspace From fb4845c38211e0475a00d1017966b3eafadc6637 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 15:41:36 +0800 Subject: [PATCH 02/13] remove cluster admin client --- pkg/sink/kafka/cluster_admin_client.go | 52 -------- pkg/sink/kafka/cluster_admin_client_mock.go | 136 -------------------- scripts/generate-mock.sh | 2 +- 3 files changed, 1 insertion(+), 189 deletions(-) delete mode 100644 pkg/sink/kafka/cluster_admin_client.go delete mode 100644 pkg/sink/kafka/cluster_admin_client_mock.go diff --git a/pkg/sink/kafka/cluster_admin_client.go b/pkg/sink/kafka/cluster_admin_client.go deleted file mode 100644 index 4f1ff36996..0000000000 --- a/pkg/sink/kafka/cluster_admin_client.go +++ /dev/null @@ -1,52 +0,0 @@ -// Copyright 2025 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package kafka - -// TopicDetail represent a topic's detail information. -type TopicDetail struct { - Name string - NumPartitions int32 - ReplicationFactor int16 -} - -// Broker represents a Kafka broker. -type Broker struct { - ID int32 -} - -// ClusterAdminClient is the administrative client for Kafka, -// which supports managing and inspecting topics, brokers, configurations and ACLs. -type ClusterAdminClient interface { - // GetAllBrokers return all brokers among the cluster - GetAllBrokers() []Broker - - // GetBrokerConfig returns the broker-level configuration and whether it exists. - GetBrokerConfig(configName string) (value string, found bool, err error) - - // GetTopicConfig returns the topic-level configuration and whether it exists. - GetTopicConfig(topicName string, configName string) (value string, found bool, err error) - - // GetTopicsMeta return all target topics' metadata - // if `ignoreTopicError` is true, ignore the topic error and return the metadata of valid topics - GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) - - // GetTopicsPartitionsNum return the number of partitions of each topic. - GetTopicsPartitionsNum(topics []string) (map[string]int32, error) - - // CreateTopic creates a new topic. - CreateTopic(detail *TopicDetail, validateOnly bool) error - - // Close shuts down the admin client. - Close() -} diff --git a/pkg/sink/kafka/cluster_admin_client_mock.go b/pkg/sink/kafka/cluster_admin_client_mock.go deleted file mode 100644 index dfeebbd773..0000000000 --- a/pkg/sink/kafka/cluster_admin_client_mock.go +++ /dev/null @@ -1,136 +0,0 @@ -// Code generated by MockGen. DO NOT EDIT. -// Source: pkg/sink/kafka/cluster_admin_client.go - -// Package kafka is a generated GoMock package. -package kafka - -import ( - reflect "reflect" - - gomock "github.com/golang/mock/gomock" -) - -// MockClusterAdminClient is a mock of ClusterAdminClient interface. -type MockClusterAdminClient struct { - ctrl *gomock.Controller - recorder *MockClusterAdminClientMockRecorder -} - -// MockClusterAdminClientMockRecorder is the mock recorder for MockClusterAdminClient. -type MockClusterAdminClientMockRecorder struct { - mock *MockClusterAdminClient -} - -// NewMockClusterAdminClient creates a new mock instance. -func NewMockClusterAdminClient(ctrl *gomock.Controller) *MockClusterAdminClient { - mock := &MockClusterAdminClient{ctrl: ctrl} - mock.recorder = &MockClusterAdminClientMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MockClusterAdminClient) EXPECT() *MockClusterAdminClientMockRecorder { - return m.recorder -} - -// Close mocks base method. -func (m *MockClusterAdminClient) Close() { - m.ctrl.T.Helper() - m.ctrl.Call(m, "Close") -} - -// Close indicates an expected call of Close. -func (mr *MockClusterAdminClientMockRecorder) Close() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockClusterAdminClient)(nil).Close)) -} - -// CreateTopic mocks base method. -func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "CreateTopic", detail, validateOnly) - ret0, _ := ret[0].(error) - return ret0 -} - -// CreateTopic indicates an expected call of CreateTopic. -func (mr *MockClusterAdminClientMockRecorder) CreateTopic(detail, validateOnly interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockClusterAdminClient)(nil).CreateTopic), detail, validateOnly) -} - -// GetAllBrokers mocks base method. -func (m *MockClusterAdminClient) GetAllBrokers() []Broker { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetAllBrokers") - ret0, _ := ret[0].([]Broker) - return ret0 -} - -// GetAllBrokers indicates an expected call of GetAllBrokers. -func (mr *MockClusterAdminClientMockRecorder) GetAllBrokers() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllBrokers", reflect.TypeOf((*MockClusterAdminClient)(nil).GetAllBrokers)) -} - -// GetBrokerConfig mocks base method. -func (m *MockClusterAdminClient) GetBrokerConfig(configName string) (string, bool, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetBrokerConfig", configName) - ret0, _ := ret[0].(string) - ret1, _ := ret[1].(bool) - ret2, _ := ret[2].(error) - return ret0, ret1, ret2 -} - -// GetBrokerConfig indicates an expected call of GetBrokerConfig. -func (mr *MockClusterAdminClientMockRecorder) GetBrokerConfig(configName interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBrokerConfig", reflect.TypeOf((*MockClusterAdminClient)(nil).GetBrokerConfig), configName) -} - -// GetTopicConfig mocks base method. -func (m *MockClusterAdminClient) GetTopicConfig(topicName, configName string) (string, bool, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetTopicConfig", topicName, configName) - ret0, _ := ret[0].(string) - ret1, _ := ret[1].(bool) - ret2, _ := ret[2].(error) - return ret0, ret1, ret2 -} - -// GetTopicConfig indicates an expected call of GetTopicConfig. -func (mr *MockClusterAdminClientMockRecorder) GetTopicConfig(topicName, configName interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicConfig", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicConfig), topicName, configName) -} - -// GetTopicsMeta mocks base method. -func (m *MockClusterAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetTopicsMeta", topics, ignoreTopicError) - ret0, _ := ret[0].(map[string]TopicDetail) - ret1, _ := ret[1].(error) - return ret0, ret1 -} - -// GetTopicsMeta indicates an expected call of GetTopicsMeta. -func (mr *MockClusterAdminClientMockRecorder) GetTopicsMeta(topics, ignoreTopicError interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsMeta", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicsMeta), topics, ignoreTopicError) -} - -// GetTopicsPartitionsNum mocks base method. -func (m *MockClusterAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetTopicsPartitionsNum", topics) - ret0, _ := ret[0].(map[string]int32) - ret1, _ := ret[1].(error) - return ret0, ret1 -} - -// GetTopicsPartitionsNum indicates an expected call of GetTopicsPartitionsNum. -func (mr *MockClusterAdminClientMockRecorder) GetTopicsPartitionsNum(topics interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsPartitionsNum", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicsPartitionsNum), topics) -} diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index 28be7895fe..e4f0a36b51 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -34,7 +34,7 @@ fi "$MOCKGEN" -source pkg/api/v2/changefeed.go -destination pkg/api/v2/mock/changefeed_mock.go -package mock "$MOCKGEN" -source pkg/api/v2/api_client.go -destination pkg/api/v2/mock/api_client_mock.go -package mock "$MOCKGEN" -source pkg/sink/codec/simple/marshaller.go -destination pkg/sink/codec/simple/mock/marshaller.go -"$MOCKGEN" -source pkg/sink/kafka/cluster_admin_client.go -destination pkg/sink/kafka/cluster_admin_client_mock.go -package kafka +"$MOCKGEN" -source pkg/sink/kafka/admin_client.go -destination pkg/sink/kafka/admin_client_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka From ffbb4fa02a5a320d1e87856471b01803d60afde8 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 15:42:56 +0800 Subject: [PATCH 03/13] add new admin error --- downstreamadapter/sink/kafka/helper.go | 2 +- downstreamadapter/sink/kafka/sink_test.go | 2 +- .../sink/topicmanager/kafka_topic_manager.go | 6 +- .../topicmanager/kafka_topic_manager_test.go | 20 +-- pkg/errors/error.go | 4 + pkg/errors/error_test.go | 5 + pkg/sink/kafka/admin.go | 30 +++- pkg/sink/kafka/admin_client.go | 53 +++++++ pkg/sink/kafka/admin_client_mock.go | 136 ++++++++++++++++++ pkg/sink/kafka/admin_test.go | 32 +++++ pkg/sink/kafka/factory.go | 4 +- pkg/sink/kafka/factory_mock.go | 6 +- pkg/sink/kafka/metrics_collector.go | 2 +- pkg/sink/kafka/options.go | 14 +- pkg/sink/kafka/options_test.go | 4 +- pkg/sink/kafka/sarama_factory.go | 6 +- 16 files changed, 292 insertions(+), 34 deletions(-) create mode 100644 pkg/sink/kafka/admin_client.go create mode 100644 pkg/sink/kafka/admin_client_mock.go diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index 37327548d4..fbbdd4cd50 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -36,7 +36,7 @@ type components struct { columnSelector *columnselector.ColumnSelectors eventRouter *eventrouter.EventRouter topicManager topicmanager.TopicManager - adminClient kafka.ClusterAdminClient + adminClient kafka.AdminClient factory kafka.Factory claimCheck *claimcheck.ClaimCheck } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index b43eb5a38d..248a935a13 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -162,7 +162,7 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, } options.Topic = topic - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, true).Return( map[string]kafka.TopicDetail{ kafkaSinkTestTopic: { diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 3db0938012..38028024cd 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -40,7 +40,7 @@ type kafkaTopicManager struct { defaultTopic string - admin kafka.ClusterAdminClient + admin kafka.AdminClient cfg *kafka.AutoCreateTopicConfig topics sync.Map @@ -54,7 +54,7 @@ func GetTopicManagerAndTryCreateTopic( changefeedID common.ChangeFeedID, topic string, topicCfg *kafka.AutoCreateTopicConfig, - adminClient kafka.ClusterAdminClient, + adminClient kafka.AdminClient, ) (TopicManager, error) { topicManager := newKafkaTopicManager( ctx, topic, changefeedID, adminClient, topicCfg, @@ -72,7 +72,7 @@ func newKafkaTopicManager( ctx context.Context, defaultTopic string, changefeedID common.ChangeFeedID, - admin kafka.ClusterAdminClient, + admin kafka.AdminClient, cfg *kafka.AutoCreateTopicConfig, ) *kafkaTopicManager { mgr := &kafkaTopicManager{ diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 3708d77668..c8f8057e11 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -28,7 +28,7 @@ import ( const kafkaTopicManagerTestTopic = "mock_topic" type mockAdminClientWithDeniedDescribe struct { - *kafka.MockClusterAdminClient + *kafka.MockAdminClient createTopicCalled bool describeCount int } @@ -41,7 +41,8 @@ func (m *mockAdminClientWithDeniedDescribe) GetTopicsMeta( if ignoreTopicError { return map[string]kafka.TopicDetail{}, nil } - return nil, sarama.ErrTopicAuthorizationFailed + return nil, errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs( + "describe-topic", topics[0]) } func (m *mockAdminClientWithDeniedDescribe) CreateTopic( @@ -53,7 +54,7 @@ func (m *mockAdminClientWithDeniedDescribe) CreateTopic( } type mockAdminClientWithDeniedCreate struct { - *kafka.MockClusterAdminClient + *kafka.MockAdminClient createTopicCalled bool describeCount int } @@ -71,14 +72,15 @@ func (m *mockAdminClientWithDeniedCreate) CreateTopic( validateOnly bool, ) error { m.createTopicCalled = true - return sarama.ErrClusterAuthorizationFailed + return errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs( + "create-topic", detail.Name) } func TestCreateTopic(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, PartitionNum: 2, @@ -191,7 +193,7 @@ func TestCreateTopicValidatesReplicationFactor(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) topic := "new-topic" gomock.InOrder( adminClient.EXPECT().GetTopicsMeta([]string{topic}, true). @@ -224,7 +226,7 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, PartitionNum: 2, @@ -273,7 +275,7 @@ func TestCreateTopicWithTopicDescribeDenied(t *testing.T) { ctrl := gomock.NewController(t) adminClient := &mockAdminClientWithDeniedDescribe{ - MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), + MockAdminClient: kafka.NewMockAdminClient(ctrl), } cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, @@ -302,7 +304,7 @@ func TestCreateTopicWithCreateDenied(t *testing.T) { ctrl := gomock.NewController(t) adminClient := &mockAdminClientWithDeniedCreate{ - MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), + MockAdminClient: kafka.NewMockAdminClient(ctrl), } cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, diff --git a/pkg/errors/error.go b/pkg/errors/error.go index 5dcb4160f3..9956d39544 100644 --- a/pkg/errors/error.go +++ b/pkg/errors/error.go @@ -143,6 +143,10 @@ var ( "kafka admin API %s failed: %s", errors.RFCCodeText("CDC:ErrKafkaAdminAPI"), ) + ErrKafkaAuthorizationFailed = errors.Normalize( + "kafka %s authorization failed: %s", + errors.RFCCodeText("CDC:ErrKafkaAuthorizationFailed"), + ) ErrPulsarInvalidTopicExpression = errors.Normalize( "invalid topic expression", errors.RFCCodeText("CDC:ErrPulsarTopicExprInvalid"), diff --git a/pkg/errors/error_test.go b/pkg/errors/error_test.go index 2c76ca8978..31879157cb 100644 --- a/pkg/errors/error_test.go +++ b/pkg/errors/error_test.go @@ -108,6 +108,11 @@ func TestShouldFailChangefeed(t *testing.T) { err: ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", "test-topic"), expected: false, }, + { + name: "ErrKafkaAuthorizationFailed should return false", + err: ErrKafkaAuthorizationFailed.GenWithStackByArgs("describe-topic", "test-topic"), + expected: false, + }, { name: "ErrKafkaSendMessage should return false", err: ErrKafkaSendMessage.GenWithStackByArgs(), diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 98e3b154d1..caf77a0ad7 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -110,6 +110,15 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool metaList, err := a.admin.DescribeTopics(topics) if err != nil { + if errors.Is(err, sarama.ErrTopicAuthorizationFailed) || + errors.Is(err, sarama.ErrClusterAuthorizationFailed) { + return nil, errors.WrapError( + errors.ErrKafkaAuthorizationFailed, + err, + "describe-topics", + strings.Join(topics, ","), + ) + } return nil, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-topics", strings.Join(topics, ",")) } @@ -119,6 +128,15 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool continue } if !ignoreTopicError { + if errors.Is(meta.Err, sarama.ErrTopicAuthorizationFailed) || + errors.Is(meta.Err, sarama.ErrClusterAuthorizationFailed) { + return nil, errors.WrapError( + errors.ErrKafkaAuthorizationFailed, + meta.Err, + "describe-topic", + meta.Name, + ) + } return nil, errors.WrapError(errors.ErrKafkaAdminAPI, meta.Err, "describe-topic", meta.Name) } log.Warn("kafka topic metadata refresh failed", @@ -138,8 +156,7 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool // IsAdminAuthorizationFailed checks whether err is an authorization failure from Kafka admin APIs. func IsAdminAuthorizationFailed(err error) bool { - return errors.Is(err, sarama.ErrTopicAuthorizationFailed) || - errors.Is(err, sarama.ErrClusterAuthorizationFailed) + return errors.Is(err, errors.ErrKafkaAuthorizationFailed) } func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { @@ -164,6 +181,15 @@ func (a *saramaAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) err := a.admin.CreateTopic(detail.Name, request, validateOnly) // Ignore the already exists error because it's not harmful. if err != nil && !strings.Contains(err.Error(), sarama.ErrTopicAlreadyExists.Error()) { + if errors.Is(err, sarama.ErrTopicAuthorizationFailed) || + errors.Is(err, sarama.ErrClusterAuthorizationFailed) { + return errors.WrapError( + errors.ErrKafkaAuthorizationFailed, + err, + "create-topic", + detail.Name, + ) + } return errors.WrapError(errors.ErrKafkaAdminAPI, err, "create-topic", detail.Name) } return nil diff --git a/pkg/sink/kafka/admin_client.go b/pkg/sink/kafka/admin_client.go new file mode 100644 index 0000000000..e6ce9f20d4 --- /dev/null +++ b/pkg/sink/kafka/admin_client.go @@ -0,0 +1,53 @@ +// Copyright 2025 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +// TopicDetail represent a topic's detail information. +type TopicDetail struct { + Name string + NumPartitions int32 + ReplicationFactor int16 +} + +// Broker represents a Kafka broker. +type Broker struct { + ID int32 +} + +// AdminClient is the administrative client for Kafka, +// which supports managing and inspecting topics, brokers, configurations and ACLs. +type AdminClient interface { + // GetAllBrokers return all brokers among the cluster + GetAllBrokers() []Broker + + // GetBrokerConfig returns the broker-level configuration and whether it exists. + GetBrokerConfig(configName string) (value string, found bool, err error) + + // GetTopicConfig returns the topic-level configuration and whether it exists. + GetTopicConfig(topicName string, configName string) (value string, found bool, err error) + + // GetTopicsMeta return all target topics' metadata + // if `ignoreTopicError` is true, ignore the topic error and return the metadata of valid topics + GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) + + // GetTopicsPartitionsNum return the number of partitions of each topic. + GetTopicsPartitionsNum(topics []string) (map[string]int32, error) + + // CreateTopic creates a new topic. + CreateTopic(detail *TopicDetail, validateOnly bool) error + + // Close shuts down the admin client. + Close() +} diff --git a/pkg/sink/kafka/admin_client_mock.go b/pkg/sink/kafka/admin_client_mock.go new file mode 100644 index 0000000000..ccf1daedda --- /dev/null +++ b/pkg/sink/kafka/admin_client_mock.go @@ -0,0 +1,136 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: pkg/sink/kafka/admin_client.go + +// Package kafka is a generated GoMock package. +package kafka + +import ( + reflect "reflect" + + gomock "github.com/golang/mock/gomock" +) + +// MockAdminClient is a mock of AdminClient interface. +type MockAdminClient struct { + ctrl *gomock.Controller + recorder *MockAdminClientMockRecorder +} + +// MockAdminClientMockRecorder is the mock recorder for MockAdminClient. +type MockAdminClientMockRecorder struct { + mock *MockAdminClient +} + +// NewMockAdminClient creates a new mock instance. +func NewMockAdminClient(ctrl *gomock.Controller) *MockAdminClient { + mock := &MockAdminClient{ctrl: ctrl} + mock.recorder = &MockAdminClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockAdminClient) EXPECT() *MockAdminClientMockRecorder { + return m.recorder +} + +// Close mocks base method. +func (m *MockAdminClient) Close() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Close") +} + +// Close indicates an expected call of Close. +func (mr *MockAdminClientMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockAdminClient)(nil).Close)) +} + +// CreateTopic mocks base method. +func (m *MockAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateTopic", detail, validateOnly) + ret0, _ := ret[0].(error) + return ret0 +} + +// CreateTopic indicates an expected call of CreateTopic. +func (mr *MockAdminClientMockRecorder) CreateTopic(detail, validateOnly interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockAdminClient)(nil).CreateTopic), detail, validateOnly) +} + +// GetAllBrokers mocks base method. +func (m *MockAdminClient) GetAllBrokers() []Broker { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetAllBrokers") + ret0, _ := ret[0].([]Broker) + return ret0 +} + +// GetAllBrokers indicates an expected call of GetAllBrokers. +func (mr *MockAdminClientMockRecorder) GetAllBrokers() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllBrokers", reflect.TypeOf((*MockAdminClient)(nil).GetAllBrokers)) +} + +// GetBrokerConfig mocks base method. +func (m *MockAdminClient) GetBrokerConfig(configName string) (string, bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetBrokerConfig", configName) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// GetBrokerConfig indicates an expected call of GetBrokerConfig. +func (mr *MockAdminClientMockRecorder) GetBrokerConfig(configName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBrokerConfig", reflect.TypeOf((*MockAdminClient)(nil).GetBrokerConfig), configName) +} + +// GetTopicConfig mocks base method. +func (m *MockAdminClient) GetTopicConfig(topicName, configName string) (string, bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicConfig", topicName, configName) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(bool) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// GetTopicConfig indicates an expected call of GetTopicConfig. +func (mr *MockAdminClientMockRecorder) GetTopicConfig(topicName, configName interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicConfig", reflect.TypeOf((*MockAdminClient)(nil).GetTopicConfig), topicName, configName) +} + +// GetTopicsMeta mocks base method. +func (m *MockAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicsMeta", topics, ignoreTopicError) + ret0, _ := ret[0].(map[string]TopicDetail) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetTopicsMeta indicates an expected call of GetTopicsMeta. +func (mr *MockAdminClientMockRecorder) GetTopicsMeta(topics, ignoreTopicError interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsMeta", reflect.TypeOf((*MockAdminClient)(nil).GetTopicsMeta), topics, ignoreTopicError) +} + +// GetTopicsPartitionsNum mocks base method. +func (m *MockAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopicsPartitionsNum", topics) + ret0, _ := ret[0].(map[string]int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetTopicsPartitionsNum indicates an expected call of GetTopicsPartitionsNum. +func (mr *MockAdminClientMockRecorder) GetTopicsPartitionsNum(topics interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsPartitionsNum", reflect.TypeOf((*MockAdminClient)(nil).GetTopicsPartitionsNum), topics) +} diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 3bcd3d0468..45f80f76eb 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -61,6 +61,38 @@ func TestGetBrokerConfig(t *testing.T) { }) } +func TestAdminAuthorizationError(t *testing.T) { + t.Parallel() + + t.Run("describe topics", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}). + Return(nil, sarama.ErrTopicAuthorizationFailed) + client := &saramaAdminClient{admin: admin} + + _, err := client.GetTopicsMeta([]string{"test-topic"}, false) + + require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) + require.ErrorIs(t, err, sarama.ErrTopicAuthorizationFailed) + require.True(t, IsAdminAuthorizationFailed(err)) + }) + + t.Run("create topic", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().CreateTopic("test-topic", gomock.Any(), false). + Return(sarama.ErrClusterAuthorizationFailed) + client := &saramaAdminClient{admin: admin} + + err := client.CreateTopic(&TopicDetail{Name: "test-topic"}, false) + + require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) + require.ErrorIs(t, err, sarama.ErrClusterAuthorizationFailed) + require.True(t, IsAdminAuthorizationFailed(err)) + }) +} + func TestAdminClientClose(t *testing.T) { tests := []struct { name string diff --git a/pkg/sink/kafka/factory.go b/pkg/sink/kafka/factory.go index 14d83b390c..c19089de4c 100644 --- a/pkg/sink/kafka/factory.go +++ b/pkg/sink/kafka/factory.go @@ -22,13 +22,13 @@ import ( // Factory is used to produce all kafka components. type Factory interface { // AdminClient return a kafka cluster admin client - AdminClient(ctx context.Context) (ClusterAdminClient, error) + AdminClient(ctx context.Context) (AdminClient, error) // SyncProducer creates a sync producer to writer message to kafka SyncProducer(ctx context.Context) (SyncProducer, error) // AsyncProducer creates an async producer to writer message to kafka AsyncProducer(ctx context.Context) (AsyncProducer, error) // MetricsCollector returns the kafka metrics collector - MetricsCollector(adminClient ClusterAdminClient) MetricsCollector + MetricsCollector(adminClient AdminClient) MetricsCollector } // SyncProducer is the kafka sync producer diff --git a/pkg/sink/kafka/factory_mock.go b/pkg/sink/kafka/factory_mock.go index d2e8358943..ecc8fe131c 100644 --- a/pkg/sink/kafka/factory_mock.go +++ b/pkg/sink/kafka/factory_mock.go @@ -36,10 +36,10 @@ func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { } // AdminClient mocks base method. -func (m *MockFactory) AdminClient(ctx context.Context) (ClusterAdminClient, error) { +func (m *MockFactory) AdminClient(ctx context.Context) (AdminClient, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "AdminClient", ctx) - ret0, _ := ret[0].(ClusterAdminClient) + ret0, _ := ret[0].(AdminClient) ret1, _ := ret[1].(error) return ret0, ret1 } @@ -66,7 +66,7 @@ func (mr *MockFactoryMockRecorder) AsyncProducer(ctx interface{}) *gomock.Call { } // MetricsCollector mocks base method. -func (m *MockFactory) MetricsCollector(adminClient ClusterAdminClient) MetricsCollector { +func (m *MockFactory) MetricsCollector(adminClient AdminClient) MetricsCollector { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "MetricsCollector", adminClient) ret0, _ := ret[0].(MetricsCollector) diff --git a/pkg/sink/kafka/metrics_collector.go b/pkg/sink/kafka/metrics_collector.go index 96f969dd39..3e54c7d82f 100644 --- a/pkg/sink/kafka/metrics_collector.go +++ b/pkg/sink/kafka/metrics_collector.go @@ -58,7 +58,7 @@ const ( type saramaMetricsCollector struct { changefeedID common.ChangeFeedID // adminClient is used to get broker infos from broker. - adminClient ClusterAdminClient + adminClient AdminClient brokers map[int32]struct{} registry metrics.Registry } diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index da1dfb3640..35ead5fcec 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -563,7 +563,7 @@ func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { // ValidateReplicationFactor checks whether a topic created with this config // can satisfy the configured acknowledgment requirement. -func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClient) error { +func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin AdminClient) error { if c.RequiredAcks != WaitForAll { return nil } @@ -627,7 +627,7 @@ func NewKafkaClientID(captureAddr string, // from the topic or broker configuration. func adjustOptions( changefeedID common.ChangeFeedID, - admin ClusterAdminClient, + admin AdminClient, options *options, topic string, ) error { @@ -654,7 +654,7 @@ func adjustOptions( func adjustExistingTopicOption( changefeedID common.ChangeFeedID, - admin ClusterAdminClient, + admin AdminClient, options *options, info TopicDetail, ) error { @@ -674,7 +674,7 @@ func adjustExistingTopicOption( } func adjustNewTopicOptions( - admin ClusterAdminClient, + admin AdminClient, changefeedID common.ChangeFeedID, options *options, ) { @@ -696,7 +696,7 @@ func adjustNewTopicOptions( } func getTopicMaxMessageBytes( - admin ClusterAdminClient, + admin AdminClient, topic string, ) (int, bool, error) { raw, found, err := getTopicConfig( @@ -717,7 +717,7 @@ func getTopicMaxMessageBytes( return maxMessageBytes, true, nil } -func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, bool, error) { +func getBrokerMaxMessageBytes(admin AdminClient) (int, bool, error) { raw, found, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName) if err != nil { return 0, false, err @@ -737,7 +737,7 @@ func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, bool, error) { // we will try to get it from the broker's configuration. // NOTICE: The configuration names of topic and broker may be different for the same configuration. func getTopicConfig( - admin ClusterAdminClient, + admin AdminClient, topicName string, topicConfigName string, brokerConfigName string, diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 025df9086b..0d5a5d1a8f 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -42,7 +42,7 @@ const ( ) type kafkaAdminFixture struct { - admin *MockClusterAdminClient + admin *MockAdminClient topics map[string]TopicDetail brokerConfig map[string]string topicConfig map[string]map[string]string @@ -53,7 +53,7 @@ func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { ctrl := gomock.NewController(t) fixture := &kafkaAdminFixture{ - admin: NewMockClusterAdminClient(ctrl), + admin: NewMockAdminClient(ctrl), topics: make(map[string]TopicDetail), brokerConfig: map[string]string{ BrokerMessageMaxBytesConfigName: mockBrokerMessageMaxBytes, diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 8f73ca70b5..e86cdc78a3 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -83,7 +83,7 @@ func NewSaramaFactory( }, nil } -func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config *sarama.Config) (ClusterAdminClient, error) { +func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config *sarama.Config) (AdminClient, error) { start := time.Now() client, err := sarama.NewClient(endpoints, config) duration := time.Since(start) @@ -119,7 +119,7 @@ func newAdminClient(changefeedID common.ChangeFeedID, endpoints []string, config }, nil } -func (f *saramaFactory) AdminClient(ctx context.Context) (ClusterAdminClient, error) { +func (f *saramaFactory) AdminClient(ctx context.Context) (AdminClient, error) { config, err := newSaramaConfig(ctx, f.option) if err != nil { return nil, err @@ -183,7 +183,7 @@ func (f *saramaFactory) AsyncProducer(ctx context.Context) (AsyncProducer, error } func (f *saramaFactory) MetricsCollector( - adminClient ClusterAdminClient, + adminClient AdminClient, ) MetricsCollector { return &saramaMetricsCollector{ changefeedID: f.changefeedID, From 683cf857a82da8e75336d3f960c1e231c5a71c63 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 15:50:02 +0800 Subject: [PATCH 04/13] revert title change --- pkg/sink/kafka/admin_client.go | 1 - 1 file changed, 1 deletion(-) diff --git a/pkg/sink/kafka/admin_client.go b/pkg/sink/kafka/admin_client.go index e6ce9f20d4..5bc95023f5 100644 --- a/pkg/sink/kafka/admin_client.go +++ b/pkg/sink/kafka/admin_client.go @@ -8,7 +8,6 @@ // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. From f588cf193310e7f17da7d5c41927239f3111788f Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 16:13:49 +0800 Subject: [PATCH 05/13] remove new error --- .../topicmanager/kafka_topic_manager_test.go | 16 +++++++--- pkg/errors/error.go | 4 --- pkg/errors/error_test.go | 5 --- pkg/sink/kafka/admin.go | 30 ++--------------- pkg/sink/kafka/admin_test.go | 32 ------------------- 5 files changed, 14 insertions(+), 73 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index c8f8057e11..7bea4fd051 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -41,8 +41,12 @@ func (m *mockAdminClientWithDeniedDescribe) GetTopicsMeta( if ignoreTopicError { return map[string]kafka.TopicDetail{}, nil } - return nil, errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs( - "describe-topic", topics[0]) + return nil, errors.WrapError( + errors.ErrKafkaAdminAPI, + sarama.ErrTopicAuthorizationFailed, + "describe-topic", + topics[0], + ) } func (m *mockAdminClientWithDeniedDescribe) CreateTopic( @@ -72,8 +76,12 @@ func (m *mockAdminClientWithDeniedCreate) CreateTopic( validateOnly bool, ) error { m.createTopicCalled = true - return errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs( - "create-topic", detail.Name) + return errors.WrapError( + errors.ErrKafkaAdminAPI, + sarama.ErrClusterAuthorizationFailed, + "create-topic", + detail.Name, + ) } func TestCreateTopic(t *testing.T) { diff --git a/pkg/errors/error.go b/pkg/errors/error.go index 9956d39544..5dcb4160f3 100644 --- a/pkg/errors/error.go +++ b/pkg/errors/error.go @@ -143,10 +143,6 @@ var ( "kafka admin API %s failed: %s", errors.RFCCodeText("CDC:ErrKafkaAdminAPI"), ) - ErrKafkaAuthorizationFailed = errors.Normalize( - "kafka %s authorization failed: %s", - errors.RFCCodeText("CDC:ErrKafkaAuthorizationFailed"), - ) ErrPulsarInvalidTopicExpression = errors.Normalize( "invalid topic expression", errors.RFCCodeText("CDC:ErrPulsarTopicExprInvalid"), diff --git a/pkg/errors/error_test.go b/pkg/errors/error_test.go index 31879157cb..2c76ca8978 100644 --- a/pkg/errors/error_test.go +++ b/pkg/errors/error_test.go @@ -108,11 +108,6 @@ func TestShouldFailChangefeed(t *testing.T) { err: ErrKafkaAdminAPI.GenWithStackByArgs("describe-topic", "test-topic"), expected: false, }, - { - name: "ErrKafkaAuthorizationFailed should return false", - err: ErrKafkaAuthorizationFailed.GenWithStackByArgs("describe-topic", "test-topic"), - expected: false, - }, { name: "ErrKafkaSendMessage should return false", err: ErrKafkaSendMessage.GenWithStackByArgs(), diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index caf77a0ad7..98e3b154d1 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -110,15 +110,6 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool metaList, err := a.admin.DescribeTopics(topics) if err != nil { - if errors.Is(err, sarama.ErrTopicAuthorizationFailed) || - errors.Is(err, sarama.ErrClusterAuthorizationFailed) { - return nil, errors.WrapError( - errors.ErrKafkaAuthorizationFailed, - err, - "describe-topics", - strings.Join(topics, ","), - ) - } return nil, errors.WrapError(errors.ErrKafkaAdminAPI, err, "describe-topics", strings.Join(topics, ",")) } @@ -128,15 +119,6 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool continue } if !ignoreTopicError { - if errors.Is(meta.Err, sarama.ErrTopicAuthorizationFailed) || - errors.Is(meta.Err, sarama.ErrClusterAuthorizationFailed) { - return nil, errors.WrapError( - errors.ErrKafkaAuthorizationFailed, - meta.Err, - "describe-topic", - meta.Name, - ) - } return nil, errors.WrapError(errors.ErrKafkaAdminAPI, meta.Err, "describe-topic", meta.Name) } log.Warn("kafka topic metadata refresh failed", @@ -156,7 +138,8 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool // IsAdminAuthorizationFailed checks whether err is an authorization failure from Kafka admin APIs. func IsAdminAuthorizationFailed(err error) bool { - return errors.Is(err, errors.ErrKafkaAuthorizationFailed) + return errors.Is(err, sarama.ErrTopicAuthorizationFailed) || + errors.Is(err, sarama.ErrClusterAuthorizationFailed) } func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { @@ -181,15 +164,6 @@ func (a *saramaAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) err := a.admin.CreateTopic(detail.Name, request, validateOnly) // Ignore the already exists error because it's not harmful. if err != nil && !strings.Contains(err.Error(), sarama.ErrTopicAlreadyExists.Error()) { - if errors.Is(err, sarama.ErrTopicAuthorizationFailed) || - errors.Is(err, sarama.ErrClusterAuthorizationFailed) { - return errors.WrapError( - errors.ErrKafkaAuthorizationFailed, - err, - "create-topic", - detail.Name, - ) - } return errors.WrapError(errors.ErrKafkaAdminAPI, err, "create-topic", detail.Name) } return nil diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 45f80f76eb..3bcd3d0468 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -61,38 +61,6 @@ func TestGetBrokerConfig(t *testing.T) { }) } -func TestAdminAuthorizationError(t *testing.T) { - t.Parallel() - - t.Run("describe topics", func(t *testing.T) { - ctrl := gomock.NewController(t) - admin := NewMocksaramaClusterAdmin(ctrl) - admin.EXPECT().DescribeTopics([]string{"test-topic"}). - Return(nil, sarama.ErrTopicAuthorizationFailed) - client := &saramaAdminClient{admin: admin} - - _, err := client.GetTopicsMeta([]string{"test-topic"}, false) - - require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) - require.ErrorIs(t, err, sarama.ErrTopicAuthorizationFailed) - require.True(t, IsAdminAuthorizationFailed(err)) - }) - - t.Run("create topic", func(t *testing.T) { - ctrl := gomock.NewController(t) - admin := NewMocksaramaClusterAdmin(ctrl) - admin.EXPECT().CreateTopic("test-topic", gomock.Any(), false). - Return(sarama.ErrClusterAuthorizationFailed) - client := &saramaAdminClient{admin: admin} - - err := client.CreateTopic(&TopicDetail{Name: "test-topic"}, false) - - require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) - require.ErrorIs(t, err, sarama.ErrClusterAuthorizationFailed) - require.True(t, IsAdminAuthorizationFailed(err)) - }) -} - func TestAdminClientClose(t *testing.T) { tests := []struct { name string From 39ee2072228c6776838059c41a8f2762d8615947 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 17:22:06 +0800 Subject: [PATCH 06/13] kafka sink verify decouple sarama --- downstreamadapter/sink/kafka/sink.go | 8 +- downstreamadapter/sink/kafka/sink_test.go | 38 +++--- pkg/sink/kafka/sarama_async_producer.go | 10 +- pkg/sink/kafka/sarama_factory.go | 2 +- pkg/sink/kafka/sarama_sync_producer.go | 73 ++++++++--- pkg/sink/kafka/sarama_sync_producer_mock.go | 130 -------------------- pkg/sink/kafka/sarama_sync_producer_test.go | 96 +++++++++------ scripts/generate-mock.sh | 1 - 8 files changed, 146 insertions(+), 212 deletions(-) delete mode 100644 pkg/sink/kafka/sarama_sync_producer_mock.go diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 2133e1c6ba..d5bd802a60 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -71,6 +71,10 @@ func (s *sink) SinkType() common.SinkType { return common.KafkaSinkType } +var createKafkaFactory = func(createSaramaFactory func() (kafka.Factory, error)) (kafka.Factory, error) { + return createSaramaFactory() +} + func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol)) if err != nil { @@ -111,7 +115,9 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL, return err } - factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) + factory, err := createKafkaFactory(func() (kafka.Factory, error) { + return kafka.NewSaramaFactory(ctx, options, changefeedID) + }) if err != nil { return err } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 248a935a13..5f827abd49 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -22,7 +22,6 @@ import ( "testing" "time" - "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" @@ -92,23 +91,6 @@ func TestSinkWorkersReturnContextError(t *testing.T) { } func TestVerifyInvalidConfig(t *testing.T) { - broker := sarama.NewMockBroker(t, 1) - defer broker.Close() - broker.SetHandlerByMap(map[string]sarama.MockResponse{ - "ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys( - []sarama.ApiVersionsResponseKey{ - {ApiKey: 0}, - {ApiKey: 1}, - {ApiKey: 2}, - {ApiKey: 3, MaxVersion: 9}, - }), - "MetadataRequest": sarama.NewMockMetadataResponse(t). - SetController(broker.BrokerID()). - SetBroker(broker.Addr(), broker.BrokerID()). - SetLeader(kafkaSinkTestTopic, 0, broker.BrokerID()), - "DescribeConfigsRequest": sarama.NewMockDescribeConfigsResponse(t), - }) - schemaRegistry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { http.Error(w, "invalid response", http.StatusInternalServerError) })) @@ -119,10 +101,28 @@ func TestVerifyInvalidConfig(t *testing.T) { Protocol: &avroProtocol, SchemaRegistry: &schemaRegistry.URL, } - sinkURI, err := url.Parse("kafka://" + broker.Addr() + "/" + kafkaSinkTestTopic + + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/" + kafkaSinkTestTopic + "?required-acks=1&kafka-version=2.4.0") require.NoError(t, err) + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + factory := kafka.NewMockFactory(ctrl) + gomock.InOrder( + factory.EXPECT().AdminClient(gomock.Any()).Return(adminClient, nil), + adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, false).Return( + map[string]kafka.TopicDetail{kafkaSinkTestTopic: {Name: kafkaSinkTestTopic}}, nil), + adminClient.EXPECT().Close(), + ) + + originalCreateKafkaFactory := createKafkaFactory + createKafkaFactory = func(_ func() (kafka.Factory, error)) (kafka.Factory, error) { + return factory, nil + } + t.Cleanup(func() { + createKafkaFactory = originalCreateKafkaFactory + }) + changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config") err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig) require.ErrorContains(t, err, "ErrAvroSchemaAPIError") diff --git a/pkg/sink/kafka/sarama_async_producer.go b/pkg/sink/kafka/sarama_async_producer.go index d3e1781c71..5a9ea691e1 100644 --- a/pkg/sink/kafka/sarama_async_producer.go +++ b/pkg/sink/kafka/sarama_async_producer.go @@ -119,19 +119,19 @@ func (p *saramaAsyncProducer) AsyncRunCallback( if err == nil { return nil } - return p.handleProducerError(err) + return p.handleProducerError(err.Err, extractLogInfo(err.Msg)) } } } -func (p *saramaAsyncProducer) handleProducerError(err *sarama.ProducerError) error { +func (p *saramaAsyncProducer) handleProducerError(err error, logInfo *codecCommon.MessageLogInfo) error { log.Error("kafka message send failed", zap.String("keyspace", p.changefeedID.Keyspace()), zap.String("changefeed", p.changefeedID.Name()), zap.String("eventContext", BuildEventLogContext( - p.changefeedID.Keyspace(), p.changefeedID.Name(), extractLogInfo(err.Msg))), - zap.Error(err.Err)) - return errors.WrapError(errors.ErrKafkaSendMessage, err.Err) + p.changefeedID.Keyspace(), p.changefeedID.Name(), logInfo)), + zap.Error(err)) + return errors.WrapError(errors.ErrKafkaSendMessage, err) } // AsyncSend is the input channel for the user to write messages to that they diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index e86cdc78a3..3490310c31 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -150,7 +150,7 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) return &saramaSyncProducer{ id: f.changefeedID, client: client, - producer: p, + producer: &saramaSyncProducerBackend{producer: p}, closed: atomic.NewBool(false), }, nil } diff --git a/pkg/sink/kafka/sarama_sync_producer.go b/pkg/sink/kafka/sarama_sync_producer.go index fcf1c9c258..60f1ba9872 100644 --- a/pkg/sink/kafka/sarama_sync_producer.go +++ b/pkg/sink/kafka/sarama_sync_producer.go @@ -14,6 +14,7 @@ package kafka import ( + "io" "time" "github.com/IBM/sarama" @@ -25,21 +26,53 @@ import ( "go.uber.org/zap" ) -type saramaSyncClient interface { - Brokers() []*sarama.Broker - Close() error +type syncProducerMessage struct { + topic string + key []byte + value []byte + partition int32 } -type saramaSyncProducerClient interface { - SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) - SendMessages(msgs []*sarama.ProducerMessage) error +type syncProducerBackend interface { + SendMessage(msg syncProducerMessage) error + SendMessages(msgs []syncProducerMessage) error Close() error } +type saramaSyncProducerBackend struct { + producer sarama.SyncProducer +} + +func (p *saramaSyncProducerBackend) SendMessage(msg syncProducerMessage) error { + _, _, err := p.producer.SendMessage(toSaramaProducerMessage(msg)) + return err +} + +func (p *saramaSyncProducerBackend) SendMessages(msgs []syncProducerMessage) error { + saramaMessages := make([]*sarama.ProducerMessage, len(msgs)) + for i, msg := range msgs { + saramaMessages[i] = toSaramaProducerMessage(msg) + } + return p.producer.SendMessages(saramaMessages) +} + +func (p *saramaSyncProducerBackend) Close() error { + return p.producer.Close() +} + +func toSaramaProducerMessage(msg syncProducerMessage) *sarama.ProducerMessage { + return &sarama.ProducerMessage{ + Topic: msg.topic, + Key: sarama.ByteEncoder(msg.key), + Value: sarama.ByteEncoder(msg.value), + Partition: msg.partition, + } +} + type saramaSyncProducer struct { id common.ChangeFeedID - client saramaSyncClient - producer saramaSyncProducerClient + client io.Closer + producer syncProducerBackend closed *atomic.Bool } @@ -48,13 +81,13 @@ func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, messa return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } - msg := &sarama.ProducerMessage{ - Topic: topic, - Key: sarama.ByteEncoder(message.Key), - Value: sarama.ByteEncoder(message.Value), - Partition: partitionNum, + msg := syncProducerMessage{ + topic: topic, + key: message.Key, + value: message.Value, + partition: partitionNum, } - _, _, err := p.producer.SendMessage(msg) + err := p.producer.SendMessage(msg) if err == nil { return nil } @@ -71,13 +104,13 @@ func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, mess return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } - msgs := make([]*sarama.ProducerMessage, partitionNum) + msgs := make([]syncProducerMessage, partitionNum) for i := 0; i < int(partitionNum); i++ { - msgs[i] = &sarama.ProducerMessage{ - Topic: topic, - Key: sarama.ByteEncoder(message.Key), - Value: sarama.ByteEncoder(message.Value), - Partition: int32(i), + msgs[i] = syncProducerMessage{ + topic: topic, + key: message.Key, + value: message.Value, + partition: int32(i), } } err := p.producer.SendMessages(msgs) diff --git a/pkg/sink/kafka/sarama_sync_producer_mock.go b/pkg/sink/kafka/sarama_sync_producer_mock.go deleted file mode 100644 index 78671e02f2..0000000000 --- a/pkg/sink/kafka/sarama_sync_producer_mock.go +++ /dev/null @@ -1,130 +0,0 @@ -// Code generated by MockGen. DO NOT EDIT. -// Source: pkg/sink/kafka/sarama_sync_producer.go - -// Package kafka is a generated GoMock package. -package kafka - -import ( - reflect "reflect" - - sarama "github.com/IBM/sarama" - gomock "github.com/golang/mock/gomock" -) - -// MocksaramaSyncClient is a mock of saramaSyncClient interface. -type MocksaramaSyncClient struct { - ctrl *gomock.Controller - recorder *MocksaramaSyncClientMockRecorder -} - -// MocksaramaSyncClientMockRecorder is the mock recorder for MocksaramaSyncClient. -type MocksaramaSyncClientMockRecorder struct { - mock *MocksaramaSyncClient -} - -// NewMocksaramaSyncClient creates a new mock instance. -func NewMocksaramaSyncClient(ctrl *gomock.Controller) *MocksaramaSyncClient { - mock := &MocksaramaSyncClient{ctrl: ctrl} - mock.recorder = &MocksaramaSyncClientMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MocksaramaSyncClient) EXPECT() *MocksaramaSyncClientMockRecorder { - return m.recorder -} - -// Brokers mocks base method. -func (m *MocksaramaSyncClient) Brokers() []*sarama.Broker { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Brokers") - ret0, _ := ret[0].([]*sarama.Broker) - return ret0 -} - -// Brokers indicates an expected call of Brokers. -func (mr *MocksaramaSyncClientMockRecorder) Brokers() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Brokers", reflect.TypeOf((*MocksaramaSyncClient)(nil).Brokers)) -} - -// Close mocks base method. -func (m *MocksaramaSyncClient) Close() error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Close") - ret0, _ := ret[0].(error) - return ret0 -} - -// Close indicates an expected call of Close. -func (mr *MocksaramaSyncClientMockRecorder) Close() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MocksaramaSyncClient)(nil).Close)) -} - -// MocksaramaSyncProducerClient is a mock of saramaSyncProducerClient interface. -type MocksaramaSyncProducerClient struct { - ctrl *gomock.Controller - recorder *MocksaramaSyncProducerClientMockRecorder -} - -// MocksaramaSyncProducerClientMockRecorder is the mock recorder for MocksaramaSyncProducerClient. -type MocksaramaSyncProducerClientMockRecorder struct { - mock *MocksaramaSyncProducerClient -} - -// NewMocksaramaSyncProducerClient creates a new mock instance. -func NewMocksaramaSyncProducerClient(ctrl *gomock.Controller) *MocksaramaSyncProducerClient { - mock := &MocksaramaSyncProducerClient{ctrl: ctrl} - mock.recorder = &MocksaramaSyncProducerClientMockRecorder{mock} - return mock -} - -// EXPECT returns an object that allows the caller to indicate expected use. -func (m *MocksaramaSyncProducerClient) EXPECT() *MocksaramaSyncProducerClientMockRecorder { - return m.recorder -} - -// Close mocks base method. -func (m *MocksaramaSyncProducerClient) Close() error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Close") - ret0, _ := ret[0].(error) - return ret0 -} - -// Close indicates an expected call of Close. -func (mr *MocksaramaSyncProducerClientMockRecorder) Close() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).Close)) -} - -// SendMessage mocks base method. -func (m *MocksaramaSyncProducerClient) SendMessage(msg *sarama.ProducerMessage) (int32, int64, error) { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "SendMessage", msg) - ret0, _ := ret[0].(int32) - ret1, _ := ret[1].(int64) - ret2, _ := ret[2].(error) - return ret0, ret1, ret2 -} - -// SendMessage indicates an expected call of SendMessage. -func (mr *MocksaramaSyncProducerClientMockRecorder) SendMessage(msg interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMessage", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).SendMessage), msg) -} - -// SendMessages mocks base method. -func (m *MocksaramaSyncProducerClient) SendMessages(msgs []*sarama.ProducerMessage) error { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "SendMessages", msgs) - ret0, _ := ret[0].(error) - return ret0 -} - -// SendMessages indicates an expected call of SendMessages. -func (mr *MocksaramaSyncProducerClientMockRecorder) SendMessages(msgs interface{}) *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMessages", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).SendMessages), msgs) -} diff --git a/pkg/sink/kafka/sarama_sync_producer_test.go b/pkg/sink/kafka/sarama_sync_producer_test.go index 522dc13bcb..6c069de40c 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -19,8 +19,6 @@ import ( "strings" "testing" - "github.com/IBM/sarama" - "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" @@ -28,6 +26,44 @@ import ( "go.uber.org/atomic" ) +type syncProducerBackendTestDouble struct { + messages []syncProducerMessage + sendErr error + closeErr error + closeCalls int + closeOrder *[]string +} + +func (p *syncProducerBackendTestDouble) SendMessage(msg syncProducerMessage) error { + p.messages = append(p.messages, msg) + return p.sendErr +} + +func (p *syncProducerBackendTestDouble) SendMessages(msgs []syncProducerMessage) error { + p.messages = append(p.messages, msgs...) + return p.sendErr +} + +func (p *syncProducerBackendTestDouble) Close() error { + p.closeCalls++ + if p.closeOrder != nil { + *p.closeOrder = append(*p.closeOrder, "producer") + } + return p.closeErr +} + +type closeTestDouble struct { + closeErr error + closeCalls int + closeOrder *[]string +} + +func (c *closeTestDouble) Close() error { + c.closeCalls++ + *c.closeOrder = append(*c.closeOrder, "client") + return c.closeErr +} + func TestProducerRejectsSendAfterClose(t *testing.T) { t.Parallel() @@ -46,7 +82,7 @@ func TestSyncProducerClose(t *testing.T) { clientCloseErr error }{ { - name: "closes client and producer", + name: "closes client and producer once in order", }, { name: "still closes producer when client close fails", @@ -56,22 +92,25 @@ func TestSyncProducerClose(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - client := NewMocksaramaSyncClient(ctrl) - producer := NewMocksaramaSyncProducerClient(ctrl) - gomock.InOrder( - client.EXPECT().Close().Return(test.clientCloseErr), - producer.EXPECT().Close().Return(nil), - ) - - p := &saramaSyncProducer{ + closeOrder := make([]string, 0, 2) + client := &closeTestDouble{ + closeErr: test.clientCloseErr, + closeOrder: &closeOrder, + } + backend := &syncProducerBackendTestDouble{closeOrder: &closeOrder} + producer := &saramaSyncProducer{ id: common.NewChangeFeedIDWithName("test", "default"), client: client, - producer: producer, + producer: backend, closed: atomic.NewBool(false), } - p.Close() + producer.Close() + producer.Close() + + require.Equal(t, []string{"client", "producer"}, closeOrder) + require.Equal(t, 1, client.closeCalls) + require.Equal(t, 1, backend.closeCalls) }) } } @@ -79,24 +118,17 @@ func TestSyncProducerClose(t *testing.T) { func TestSyncProducerErrorWrappedOnce(t *testing.T) { cause := io.ErrClosedPipe tests := []struct { - name string - expectSend func(*MocksaramaSyncProducerClient) - send func(*saramaSyncProducer, *codecCommon.Message) error + name string + send func(*saramaSyncProducer, *codecCommon.Message) error }{ { name: "single message", - expectSend: func(producer *MocksaramaSyncProducerClient) { - producer.EXPECT().SendMessage(gomock.Any()).Return(int32(0), int64(0), cause) - }, send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { return producer.SendMessage("topic", 0, message) }, }, { name: "message batch", - expectSend: func(producer *MocksaramaSyncProducerClient) { - producer.EXPECT().SendMessages(gomock.Any()).Return(cause) - }, send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { return producer.SendMessages("topic", 1, message) }, @@ -105,17 +137,15 @@ func TestSyncProducerErrorWrappedOnce(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - ctrl := gomock.NewController(t) - producer := NewMocksaramaSyncProducerClient(ctrl) - test.expectSend(producer) - p := &saramaSyncProducer{ + backend := &syncProducerBackendTestDouble{sendErr: cause} + producer := &saramaSyncProducer{ id: common.NewChangeFeedIDWithName("test", "default"), - producer: producer, + producer: backend, closed: atomic.NewBool(false), } message := &codecCommon.Message{LogInfo: &codecCommon.MessageLogInfo{}} - err := test.send(p, message) + err := test.send(producer, message) requireKafkaSendError(t, err, cause) }) @@ -127,12 +157,8 @@ func TestAsyncProducerErrorWrappedOnce(t *testing.T) { producer := &saramaAsyncProducer{ changefeedID: common.NewChangeFeedIDWithName("test", "default"), } - err := producer.handleProducerError(&sarama.ProducerError{ - Err: cause, - Msg: &sarama.ProducerMessage{Metadata: &messageMetadata{ - logInfo: &codecCommon.MessageLogInfo{}, - }}, - }) + + err := producer.handleProducerError(cause, &codecCommon.MessageLogInfo{}) requireKafkaSendError(t, err, cause) } diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index e4f0a36b51..1f188b1527 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -37,7 +37,6 @@ fi "$MOCKGEN" -source pkg/sink/kafka/admin_client.go -destination pkg/sink/kafka/admin_client_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka -"$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka "$MOCKGEN" -source pkg/keyspace/keyspace_manager.go -destination pkg/keyspace/keyspace_manager_mock.go -package keyspace "$MOCKGEN" -source pkg/txnutil/gc/gc_manager.go -destination pkg/txnutil/gc/gc_manager_mock.go -package gc "$MOCKGEN" -source pkg/txnutil/gc/gc_client.go -destination pkg/txnutil/gc/gc_client_mock.go -package gc From 7468183b8c7bf73536827dc36a4b45d3a65eef8a Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 17:28:04 +0800 Subject: [PATCH 07/13] revert some changes --- pkg/sink/kafka/sarama_factory.go | 2 +- pkg/sink/kafka/sarama_sync_producer.go | 73 +++-------- pkg/sink/kafka/sarama_sync_producer_mock.go | 130 ++++++++++++++++++++ pkg/sink/kafka/sarama_sync_producer_test.go | 87 +++++-------- scripts/generate-mock.sh | 1 + 5 files changed, 180 insertions(+), 113 deletions(-) create mode 100644 pkg/sink/kafka/sarama_sync_producer_mock.go diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 3490310c31..e86cdc78a3 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -150,7 +150,7 @@ func (f *saramaFactory) SyncProducer(ctx context.Context) (SyncProducer, error) return &saramaSyncProducer{ id: f.changefeedID, client: client, - producer: &saramaSyncProducerBackend{producer: p}, + producer: p, closed: atomic.NewBool(false), }, nil } diff --git a/pkg/sink/kafka/sarama_sync_producer.go b/pkg/sink/kafka/sarama_sync_producer.go index 60f1ba9872..fcf1c9c258 100644 --- a/pkg/sink/kafka/sarama_sync_producer.go +++ b/pkg/sink/kafka/sarama_sync_producer.go @@ -14,7 +14,6 @@ package kafka import ( - "io" "time" "github.com/IBM/sarama" @@ -26,53 +25,21 @@ import ( "go.uber.org/zap" ) -type syncProducerMessage struct { - topic string - key []byte - value []byte - partition int32 -} - -type syncProducerBackend interface { - SendMessage(msg syncProducerMessage) error - SendMessages(msgs []syncProducerMessage) error +type saramaSyncClient interface { + Brokers() []*sarama.Broker Close() error } -type saramaSyncProducerBackend struct { - producer sarama.SyncProducer -} - -func (p *saramaSyncProducerBackend) SendMessage(msg syncProducerMessage) error { - _, _, err := p.producer.SendMessage(toSaramaProducerMessage(msg)) - return err -} - -func (p *saramaSyncProducerBackend) SendMessages(msgs []syncProducerMessage) error { - saramaMessages := make([]*sarama.ProducerMessage, len(msgs)) - for i, msg := range msgs { - saramaMessages[i] = toSaramaProducerMessage(msg) - } - return p.producer.SendMessages(saramaMessages) -} - -func (p *saramaSyncProducerBackend) Close() error { - return p.producer.Close() -} - -func toSaramaProducerMessage(msg syncProducerMessage) *sarama.ProducerMessage { - return &sarama.ProducerMessage{ - Topic: msg.topic, - Key: sarama.ByteEncoder(msg.key), - Value: sarama.ByteEncoder(msg.value), - Partition: msg.partition, - } +type saramaSyncProducerClient interface { + SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) + SendMessages(msgs []*sarama.ProducerMessage) error + Close() error } type saramaSyncProducer struct { id common.ChangeFeedID - client io.Closer - producer syncProducerBackend + client saramaSyncClient + producer saramaSyncProducerClient closed *atomic.Bool } @@ -81,13 +48,13 @@ func (p *saramaSyncProducer) SendMessage(topic string, partitionNum int32, messa return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } - msg := syncProducerMessage{ - topic: topic, - key: message.Key, - value: message.Value, - partition: partitionNum, + msg := &sarama.ProducerMessage{ + Topic: topic, + Key: sarama.ByteEncoder(message.Key), + Value: sarama.ByteEncoder(message.Value), + Partition: partitionNum, } - err := p.producer.SendMessage(msg) + _, _, err := p.producer.SendMessage(msg) if err == nil { return nil } @@ -104,13 +71,13 @@ func (p *saramaSyncProducer) SendMessages(topic string, partitionNum int32, mess return errors.ErrKafkaSinkClosed.GenWithStackByArgs() } - msgs := make([]syncProducerMessage, partitionNum) + msgs := make([]*sarama.ProducerMessage, partitionNum) for i := 0; i < int(partitionNum); i++ { - msgs[i] = syncProducerMessage{ - topic: topic, - key: message.Key, - value: message.Value, - partition: int32(i), + msgs[i] = &sarama.ProducerMessage{ + Topic: topic, + Key: sarama.ByteEncoder(message.Key), + Value: sarama.ByteEncoder(message.Value), + Partition: int32(i), } } err := p.producer.SendMessages(msgs) diff --git a/pkg/sink/kafka/sarama_sync_producer_mock.go b/pkg/sink/kafka/sarama_sync_producer_mock.go new file mode 100644 index 0000000000..78671e02f2 --- /dev/null +++ b/pkg/sink/kafka/sarama_sync_producer_mock.go @@ -0,0 +1,130 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: pkg/sink/kafka/sarama_sync_producer.go + +// Package kafka is a generated GoMock package. +package kafka + +import ( + reflect "reflect" + + sarama "github.com/IBM/sarama" + gomock "github.com/golang/mock/gomock" +) + +// MocksaramaSyncClient is a mock of saramaSyncClient interface. +type MocksaramaSyncClient struct { + ctrl *gomock.Controller + recorder *MocksaramaSyncClientMockRecorder +} + +// MocksaramaSyncClientMockRecorder is the mock recorder for MocksaramaSyncClient. +type MocksaramaSyncClientMockRecorder struct { + mock *MocksaramaSyncClient +} + +// NewMocksaramaSyncClient creates a new mock instance. +func NewMocksaramaSyncClient(ctrl *gomock.Controller) *MocksaramaSyncClient { + mock := &MocksaramaSyncClient{ctrl: ctrl} + mock.recorder = &MocksaramaSyncClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MocksaramaSyncClient) EXPECT() *MocksaramaSyncClientMockRecorder { + return m.recorder +} + +// Brokers mocks base method. +func (m *MocksaramaSyncClient) Brokers() []*sarama.Broker { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Brokers") + ret0, _ := ret[0].([]*sarama.Broker) + return ret0 +} + +// Brokers indicates an expected call of Brokers. +func (mr *MocksaramaSyncClientMockRecorder) Brokers() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Brokers", reflect.TypeOf((*MocksaramaSyncClient)(nil).Brokers)) +} + +// Close mocks base method. +func (m *MocksaramaSyncClient) Close() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Close") + ret0, _ := ret[0].(error) + return ret0 +} + +// Close indicates an expected call of Close. +func (mr *MocksaramaSyncClientMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MocksaramaSyncClient)(nil).Close)) +} + +// MocksaramaSyncProducerClient is a mock of saramaSyncProducerClient interface. +type MocksaramaSyncProducerClient struct { + ctrl *gomock.Controller + recorder *MocksaramaSyncProducerClientMockRecorder +} + +// MocksaramaSyncProducerClientMockRecorder is the mock recorder for MocksaramaSyncProducerClient. +type MocksaramaSyncProducerClientMockRecorder struct { + mock *MocksaramaSyncProducerClient +} + +// NewMocksaramaSyncProducerClient creates a new mock instance. +func NewMocksaramaSyncProducerClient(ctrl *gomock.Controller) *MocksaramaSyncProducerClient { + mock := &MocksaramaSyncProducerClient{ctrl: ctrl} + mock.recorder = &MocksaramaSyncProducerClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MocksaramaSyncProducerClient) EXPECT() *MocksaramaSyncProducerClientMockRecorder { + return m.recorder +} + +// Close mocks base method. +func (m *MocksaramaSyncProducerClient) Close() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Close") + ret0, _ := ret[0].(error) + return ret0 +} + +// Close indicates an expected call of Close. +func (mr *MocksaramaSyncProducerClientMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).Close)) +} + +// SendMessage mocks base method. +func (m *MocksaramaSyncProducerClient) SendMessage(msg *sarama.ProducerMessage) (int32, int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SendMessage", msg) + ret0, _ := ret[0].(int32) + ret1, _ := ret[1].(int64) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// SendMessage indicates an expected call of SendMessage. +func (mr *MocksaramaSyncProducerClientMockRecorder) SendMessage(msg interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMessage", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).SendMessage), msg) +} + +// SendMessages mocks base method. +func (m *MocksaramaSyncProducerClient) SendMessages(msgs []*sarama.ProducerMessage) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SendMessages", msgs) + ret0, _ := ret[0].(error) + return ret0 +} + +// SendMessages indicates an expected call of SendMessages. +func (mr *MocksaramaSyncProducerClientMockRecorder) SendMessages(msgs interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMessages", reflect.TypeOf((*MocksaramaSyncProducerClient)(nil).SendMessages), msgs) +} diff --git a/pkg/sink/kafka/sarama_sync_producer_test.go b/pkg/sink/kafka/sarama_sync_producer_test.go index 6c069de40c..e00e0dafa8 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -19,6 +19,7 @@ import ( "strings" "testing" + "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/errors" codecCommon "github.com/pingcap/ticdc/pkg/sink/codec/common" @@ -26,44 +27,6 @@ import ( "go.uber.org/atomic" ) -type syncProducerBackendTestDouble struct { - messages []syncProducerMessage - sendErr error - closeErr error - closeCalls int - closeOrder *[]string -} - -func (p *syncProducerBackendTestDouble) SendMessage(msg syncProducerMessage) error { - p.messages = append(p.messages, msg) - return p.sendErr -} - -func (p *syncProducerBackendTestDouble) SendMessages(msgs []syncProducerMessage) error { - p.messages = append(p.messages, msgs...) - return p.sendErr -} - -func (p *syncProducerBackendTestDouble) Close() error { - p.closeCalls++ - if p.closeOrder != nil { - *p.closeOrder = append(*p.closeOrder, "producer") - } - return p.closeErr -} - -type closeTestDouble struct { - closeErr error - closeCalls int - closeOrder *[]string -} - -func (c *closeTestDouble) Close() error { - c.closeCalls++ - *c.closeOrder = append(*c.closeOrder, "client") - return c.closeErr -} - func TestProducerRejectsSendAfterClose(t *testing.T) { t.Parallel() @@ -82,7 +45,7 @@ func TestSyncProducerClose(t *testing.T) { clientCloseErr error }{ { - name: "closes client and producer once in order", + name: "closes client and producer", }, { name: "still closes producer when client close fails", @@ -92,25 +55,22 @@ func TestSyncProducerClose(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - closeOrder := make([]string, 0, 2) - client := &closeTestDouble{ - closeErr: test.clientCloseErr, - closeOrder: &closeOrder, - } - backend := &syncProducerBackendTestDouble{closeOrder: &closeOrder} - producer := &saramaSyncProducer{ + ctrl := gomock.NewController(t) + client := NewMocksaramaSyncClient(ctrl) + producer := NewMocksaramaSyncProducerClient(ctrl) + gomock.InOrder( + client.EXPECT().Close().Return(test.clientCloseErr), + producer.EXPECT().Close().Return(nil), + ) + + p := &saramaSyncProducer{ id: common.NewChangeFeedIDWithName("test", "default"), client: client, - producer: backend, + producer: producer, closed: atomic.NewBool(false), } - producer.Close() - producer.Close() - - require.Equal(t, []string{"client", "producer"}, closeOrder) - require.Equal(t, 1, client.closeCalls) - require.Equal(t, 1, backend.closeCalls) + p.Close() }) } } @@ -118,17 +78,24 @@ func TestSyncProducerClose(t *testing.T) { func TestSyncProducerErrorWrappedOnce(t *testing.T) { cause := io.ErrClosedPipe tests := []struct { - name string - send func(*saramaSyncProducer, *codecCommon.Message) error + name string + expectSend func(*MocksaramaSyncProducerClient) + send func(*saramaSyncProducer, *codecCommon.Message) error }{ { name: "single message", + expectSend: func(producer *MocksaramaSyncProducerClient) { + producer.EXPECT().SendMessage(gomock.Any()).Return(int32(0), int64(0), cause) + }, send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { return producer.SendMessage("topic", 0, message) }, }, { name: "message batch", + expectSend: func(producer *MocksaramaSyncProducerClient) { + producer.EXPECT().SendMessages(gomock.Any()).Return(cause) + }, send: func(producer *saramaSyncProducer, message *codecCommon.Message) error { return producer.SendMessages("topic", 1, message) }, @@ -137,15 +104,17 @@ func TestSyncProducerErrorWrappedOnce(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - backend := &syncProducerBackendTestDouble{sendErr: cause} - producer := &saramaSyncProducer{ + ctrl := gomock.NewController(t) + producer := NewMocksaramaSyncProducerClient(ctrl) + test.expectSend(producer) + p := &saramaSyncProducer{ id: common.NewChangeFeedIDWithName("test", "default"), - producer: backend, + producer: producer, closed: atomic.NewBool(false), } message := &codecCommon.Message{LogInfo: &codecCommon.MessageLogInfo{}} - err := test.send(producer, message) + err := test.send(p, message) requireKafkaSendError(t, err, cause) }) diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh index 1f188b1527..e4f0a36b51 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -37,6 +37,7 @@ fi "$MOCKGEN" -source pkg/sink/kafka/admin_client.go -destination pkg/sink/kafka/admin_client_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/factory.go -destination pkg/sink/kafka/factory_mock.go -package kafka "$MOCKGEN" -source pkg/sink/kafka/admin.go -destination pkg/sink/kafka/admin_mock.go -package kafka +"$MOCKGEN" -source pkg/sink/kafka/sarama_sync_producer.go -destination pkg/sink/kafka/sarama_sync_producer_mock.go -package kafka "$MOCKGEN" -source pkg/keyspace/keyspace_manager.go -destination pkg/keyspace/keyspace_manager_mock.go -package keyspace "$MOCKGEN" -source pkg/txnutil/gc/gc_manager.go -destination pkg/txnutil/gc/gc_manager_mock.go -package gc "$MOCKGEN" -source pkg/txnutil/gc/gc_client.go -destination pkg/txnutil/gc/gc_client_mock.go -package gc From 1e1102dec1276a88e0aca23049a2a0afec89118f Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 17:58:45 +0800 Subject: [PATCH 08/13] remove kafka admin fixture --- pkg/sink/kafka/options_test.go | 260 +++++++++------------------------ 1 file changed, 68 insertions(+), 192 deletions(-) diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 0d5a5d1a8f..a842a2a1d3 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -14,7 +14,6 @@ package kafka import ( - "context" "fmt" "net/url" "strconv" @@ -22,7 +21,6 @@ import ( "testing" "time" - "github.com/IBM/sarama" "github.com/aws/aws-sdk-go-v2/aws" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" @@ -35,134 +33,10 @@ const ( defaultMockTopicName = "mock_topic" // These values model Kafka admin responses, not TiCDC option defaults. - mockClusterReplicationFactor int16 = 3 - mockBrokerMessageMaxBytes = "1048588" - mockTopicMessageMaxBytes = "1048588" - mockMinInsyncReplicas = "1" + mockBrokerMessageMaxBytes = "1048588" + mockTopicMessageMaxBytes = "1048588" ) -type kafkaAdminFixture struct { - admin *MockAdminClient - topics map[string]TopicDetail - brokerConfig map[string]string - topicConfig map[string]map[string]string -} - -func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { - t.Helper() - - ctrl := gomock.NewController(t) - fixture := &kafkaAdminFixture{ - admin: NewMockAdminClient(ctrl), - topics: make(map[string]TopicDetail), - brokerConfig: map[string]string{ - BrokerMessageMaxBytesConfigName: mockBrokerMessageMaxBytes, - MinInsyncReplicasConfigName: mockMinInsyncReplicas, - }, - topicConfig: make(map[string]map[string]string), - } - fixture.addTopic(defaultMockTopicName, defaultPartitionNum) - fixture.topicConfig[defaultMockTopicName] = map[string]string{ - TopicMaxMessageBytesConfigName: mockTopicMessageMaxBytes, - MinInsyncReplicasConfigName: mockMinInsyncReplicas, - } - - fixture.admin.EXPECT().Close().AnyTimes() - fixture.admin.EXPECT().GetTopicsMeta(gomock.Any(), gomock.Any()). - DoAndReturn(fixture.getTopicsMeta).AnyTimes() - fixture.admin.EXPECT().GetTopicsPartitionsNum(gomock.Any()). - DoAndReturn(fixture.getTopicsPartitionsNum).AnyTimes() - fixture.admin.EXPECT().GetBrokerConfig(gomock.Any()). - DoAndReturn(fixture.getBrokerConfig).AnyTimes() - fixture.admin.EXPECT().GetTopicConfig(gomock.Any(), gomock.Any()). - DoAndReturn(fixture.getTopicConfig).AnyTimes() - fixture.admin.EXPECT().CreateTopic(gomock.Any(), gomock.Any()). - DoAndReturn(fixture.createTopic).AnyTimes() - - return fixture -} - -func (f *kafkaAdminFixture) addTopic(name string, partitionNum int32) { - f.topics[name] = TopicDetail{Name: name, NumPartitions: partitionNum} -} - -func (f *kafkaAdminFixture) getTopicsMeta( - topics []string, _ bool, -) (map[string]TopicDetail, error) { - result := make(map[string]TopicDetail, len(topics)) - for _, topic := range topics { - if detail, ok := f.topics[topic]; ok { - result[topic] = detail - } - } - return result, nil -} - -func (f *kafkaAdminFixture) getTopicsPartitionsNum( - topics []string, -) (map[string]int32, error) { - result := make(map[string]int32, len(topics)) - for _, topic := range topics { - if detail, ok := f.topics[topic]; ok { - result[topic] = detail.NumPartitions - } - } - return result, nil -} - -func (f *kafkaAdminFixture) getBrokerConfig(configName string) (string, bool, error) { - if value, ok := f.brokerConfig[configName]; ok { - return value, true, nil - } - return "", false, nil -} - -func (f *kafkaAdminFixture) getTopicConfig(topicName string, configName string) (string, bool, error) { - if _, ok := f.topics[topicName]; !ok { - return "", false, nil - } - if value, ok := f.topicConfig[topicName][configName]; ok { - return value, true, nil - } - return "", false, nil -} - -func (f *kafkaAdminFixture) createTopic(detail *TopicDetail, _ bool) error { - if detail.ReplicationFactor > mockClusterReplicationFactor { - return sarama.ErrInvalidReplicationFactor - } - if _, ok := f.brokerConfig[MinInsyncReplicasConfigName]; !ok && - detail.ReplicationFactor != mockClusterReplicationFactor { - return sarama.ErrPolicyViolation - } - f.topics[detail.Name] = *detail - return nil -} - -func (f *kafkaAdminFixture) brokerMessageMaxBytes() int { - value, _ := strconv.Atoi(f.brokerConfig[BrokerMessageMaxBytesConfigName]) - return value -} - -func (f *kafkaAdminFixture) topicMaxMessageBytes(topicName string) int { - value, _ := strconv.Atoi(f.topicConfig[topicName][TopicMaxMessageBytesConfigName]) - return value -} - -func (f *kafkaAdminFixture) setMessageMaxBytes(brokerValue, topicValue string) { - f.brokerConfig[BrokerMessageMaxBytesConfigName] = brokerValue - f.topicConfig[defaultMockTopicName][TopicMaxMessageBytesConfigName] = topicValue -} - -func (f *kafkaAdminFixture) setMinInsyncReplicas(minInsyncReplicas string) { - f.topicConfig[defaultMockTopicName][MinInsyncReplicasConfigName] = minInsyncReplicas - f.brokerConfig[MinInsyncReplicasConfigName] = minInsyncReplicas -} - -func (f *kafkaAdminFixture) dropBrokerConfig(configName string) { - delete(f.brokerConfig, configName) -} - func TestCompleteOptions(t *testing.T) { options := NewOptions() @@ -451,27 +325,24 @@ func TestApplyRejectsNonPositiveTimeout(t *testing.T) { } func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *testing.T) { + brokerMessageMaxBytes, err := strconv.Atoi(mockBrokerMessageMaxBytes) + require.NoError(t, err) + tests := []struct { name string - configuredMaxMessageBytes func(*kafkaAdminFixture) int + configuredMaxMessageBytes int }{ { - name: "uses broker limit when configured value is below broker", - configuredMaxMessageBytes: func(*kafkaAdminFixture) int { - return 1024 - }, + name: "uses broker limit when configured value is below broker", + configuredMaxMessageBytes: 1024, }, { - name: "uses broker limit when configured value is below broker by one byte", - configuredMaxMessageBytes: func(f *kafkaAdminFixture) int { - return f.brokerMessageMaxBytes() - 1 - }, + name: "uses broker limit when configured value is below broker by one byte", + configuredMaxMessageBytes: brokerMessageMaxBytes - 1, }, { - name: "uses broker limit when configured value is above broker", - configuredMaxMessageBytes: func(f *kafkaAdminFixture) int { - return f.brokerMessageMaxBytes() + 1 - }, + name: "uses broker limit when configured value is above broker", + configuredMaxMessageBytes: brokerMessageMaxBytes + 1, }, } @@ -479,53 +350,54 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") for _, test := range tests { t.Run(test.name, func(t *testing.T) { - adminFixture := newKafkaAdminFixture(t) - adminClient := adminFixture.admin - - detail := &TopicDetail{ - Name: topicName, - NumPartitions: 3, - } - err := adminClient.CreateTopic(detail, false) - require.NoError(t, err) + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + gomock.InOrder( + adminClient.EXPECT().GetTopicsMeta([]string{topicName}, true).Return( + map[string]TopicDetail{ + topicName: {Name: topicName, NumPartitions: 3}, + }, nil), + adminClient.EXPECT().GetTopicConfig(topicName, TopicMaxMessageBytesConfigName). + Return("", false, nil), + adminClient.EXPECT().GetBrokerConfig(BrokerMessageMaxBytesConfigName). + Return(mockBrokerMessageMaxBytes, true, nil), + ) - configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture) sinkURI, err := url.Parse(fmt.Sprintf( "kafka://127.0.0.1:9092/%s?max-message-bytes=%d", - topicName, configuredMaxMessageBytes, + topicName, test.configuredMaxMessageBytes, )) require.NoError(t, err) options := NewOptions() err = options.Apply(changefeedID, sinkURI, config.GetDefaultReplicaConfig().Sink) require.NoError(t, err) - require.Equal(t, configuredMaxMessageBytes, options.MaxMessageBytes) - require.Equal(t, configuredMaxMessageBytes, options.MaxBatchedBytes) - expectedProducerLimit := adminFixture.brokerMessageMaxBytes() + require.Equal(t, test.configuredMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, test.configuredMaxMessageBytes, options.MaxBatchedBytes) - ctx := context.Background() err = adjustOptions(changefeedID, adminClient, options, topicName) require.NoError(t, err) - saramaConfig, err := newSaramaConfig(ctx, options) - require.NoError(t, err) - - require.NotEqual(t, configuredMaxMessageBytes, options.MaxMessageBytes) - require.Equal(t, expectedProducerLimit, options.MaxMessageBytes) + require.NotEqual(t, test.configuredMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, brokerMessageMaxBytes, options.MaxMessageBytes) require.Equal( t, - min(configuredMaxMessageBytes, expectedProducerLimit), + min(test.configuredMaxMessageBytes, brokerMessageMaxBytes), options.MaxBatchedBytes, ) - require.Equal(t, expectedProducerLimit, saramaConfig.Producer.MaxMessageBytes) }) } } func TestValidateReplicationFactor(t *testing.T) { - adminFixture := newKafkaAdminFixture(t) - adminClient := adminFixture.admin - adminFixture.setMinInsyncReplicas("2") + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + gomock.InOrder( + adminClient.EXPECT().GetBrokerConfig(MinInsyncReplicasConfigName). + Return("2", true, nil), + adminClient.EXPECT().GetBrokerConfig(MinInsyncReplicasConfigName). + Return("", false, nil), + ) topicConfig := &AutoCreateTopicConfig{ AutoCreate: true, @@ -547,7 +419,6 @@ func TestValidateReplicationFactor(t *testing.T) { err = localAcksConfig.ValidateReplicationFactor(adminClient) require.NoError(t, err) - adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName) missingBrokerConfig := &AutoCreateTopicConfig{ AutoCreate: true, ReplicationFactor: 1, @@ -557,15 +428,6 @@ func TestValidateReplicationFactor(t *testing.T) { require.NoError(t, err) } -func TestCreateProducerFailed(t *testing.T) { - options := NewOptions() - options.Version = "invalid" - options.IsAssignedVersion = true - saramaConfig, err := newSaramaConfig(context.Background(), options) - require.Regexp(t, "invalid version.*", errors.Cause(err)) - require.Nil(t, saramaConfig) -} - func TestConfigurationCombinations(t *testing.T) { combinations := []struct { name string @@ -716,39 +578,53 @@ func TestConfigurationCombinations(t *testing.T) { for _, a := range combinations { t.Run(a.name, func(t *testing.T) { - adminFixture := newKafkaAdminFixture(t) - adminFixture.setMessageMaxBytes(a.brokerMessageMaxBytes, a.topicMaxMessageBytes) - adminClient := adminFixture.admin - uri := fmt.Sprintf(a.uriTemplate, a.uriParams...) sinkURI, err := url.Parse(uri) require.Nil(t, err) - options := NewOptions() - err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) - require.Nil(t, err) - configuredMaxMessageBytes := options.MaxMessageBytes - topic, ok := a.uriParams[0].(string) require.True(t, ok) require.NotEqual(t, "", topic) - sourceMaxMessageBytes := adminFixture.brokerMessageMaxBytes() - if _, exists := adminFixture.topics[topic]; exists { - sourceMaxMessageBytes = adminFixture.topicMaxMessageBytes(topic) + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + metadataCall := adminClient.EXPECT().GetTopicsMeta([]string{topic}, true) + sourceMaxMessageBytes := a.brokerMessageMaxBytes + if topic == defaultMockTopicName { + metadataCall.Return(map[string]TopicDetail{ + topic: {Name: topic, NumPartitions: defaultPartitionNum}, + }, nil) + gomock.InOrder( + metadataCall, + adminClient.EXPECT().GetTopicConfig(topic, TopicMaxMessageBytesConfigName). + Return(a.topicMaxMessageBytes, true, nil), + ) + sourceMaxMessageBytes = a.topicMaxMessageBytes + } else { + metadataCall.Return(map[string]TopicDetail{}, nil) + gomock.InOrder( + metadataCall, + adminClient.EXPECT().GetBrokerConfig(BrokerMessageMaxBytesConfigName). + Return(a.brokerMessageMaxBytes, true, nil), + ) } + options := NewOptions() + err = options.Apply(common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) + require.Nil(t, err) + configuredMaxMessageBytes := options.MaxMessageBytes + + expectedMaxMessageBytes, err := strconv.Atoi(sourceMaxMessageBytes) + require.NoError(t, err) changefeedID := common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test") err = adjustOptions(changefeedID, adminClient, options, topic) require.Nil(t, err) - require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) + require.Equal(t, expectedMaxMessageBytes, options.MaxMessageBytes) require.Equal( t, - min(configuredMaxMessageBytes, sourceMaxMessageBytes), + min(configuredMaxMessageBytes, expectedMaxMessageBytes), options.MaxBatchedBytes, ) - - adminClient.Close() }) } } From cd1afecc4a20efeef71dcc30cc31d4f423248e7a Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 18:08:37 +0800 Subject: [PATCH 09/13] add tests --- pkg/sink/kafka/options_test.go | 49 ++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index a842a2a1d3..91b17dd15d 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -426,6 +426,55 @@ func TestValidateReplicationFactor(t *testing.T) { } err = missingBrokerConfig.ValidateReplicationFactor(adminClient) require.NoError(t, err) + + t.Run("replication factor satisfies min insync replicas", func(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + adminClient.EXPECT().GetBrokerConfig(MinInsyncReplicasConfigName). + Return("2", true, nil) + + topicConfig := &AutoCreateTopicConfig{ + ReplicationFactor: 3, + RequiredAcks: WaitForAll, + } + + err := topicConfig.ValidateReplicationFactor(adminClient) + require.NoError(t, err) + }) + + t.Run("invalid min insync replicas", func(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + adminClient.EXPECT().GetBrokerConfig(MinInsyncReplicasConfigName). + Return("invalid", true, nil) + + topicConfig := &AutoCreateTopicConfig{ + ReplicationFactor: 3, + RequiredAcks: WaitForAll, + } + + err := topicConfig.ValidateReplicationFactor(adminClient) + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + }) + + t.Run("broker config lookup failure skips validation", func(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := NewMockAdminClient(ctrl) + lookupErr := errors.ErrKafkaAdminAPI.GenWithStackByArgs( + "describe-config", + MinInsyncReplicasConfigName, + ) + adminClient.EXPECT().GetBrokerConfig(MinInsyncReplicasConfigName). + Return("", false, lookupErr) + + topicConfig := &AutoCreateTopicConfig{ + ReplicationFactor: 1, + RequiredAcks: WaitForAll, + } + + err := topicConfig.ValidateReplicationFactor(adminClient) + require.NoError(t, err) + }) } func TestConfigurationCombinations(t *testing.T) { From 1f6723627cfae392a42123de6ce4167ff33ff1b4 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 18:19:21 +0800 Subject: [PATCH 10/13] fix option apply tests --- pkg/sink/kafka/options_test.go | 244 ++++++++++++++++++++++++ pkg/sink/kafka/sarama_config_test.go | 267 --------------------------- 2 files changed, 244 insertions(+), 267 deletions(-) diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 91b17dd15d..6b3860fa1f 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -26,6 +26,7 @@ import ( "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/security" "github.com/stretchr/testify/require" ) @@ -164,6 +165,249 @@ func TestCompleteOptions(t *testing.T) { require.Equal(t, defaultMaxRetry, options.MaxRetry) } +func TestApplySASL(t *testing.T) { + t.Parallel() + + const baseURI = "kafka://127.0.0.1:9092/abc" + tests := []struct { + name string + uri string + kafkaConfig *config.KafkaConfig + expected security.SASL + expectErr string + }{ + {name: "no params", uri: baseURI}, + { + name: "valid PLAIN SASL", + uri: baseURI + "?sasl-user=user&sasl-password=password&sasl-mechanism=plain", + expected: security.SASL{ + SASLUser: "user", + SASLPassword: "password", + SASLMechanism: security.PlainMechanism, + }, + }, + { + name: "valid SCRAM SASL", + uri: baseURI + "?sasl-user=user&sasl-password=password&sasl-mechanism=SCRAM-SHA-512", + expected: security.SASL{ + SASLUser: "user", + SASLPassword: "password", + SASLMechanism: security.SCRAM512Mechanism, + }, + }, + { + name: "valid GSSAPI user auth SASL", + uri: baseURI + "?sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=USER" + + "&sasl-gssapi-kerberos-config-path=/root/config" + + "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + + "&sasl-gssapi-password=pwd&sasl-gssapi-realm=realm" + + "&sasl-gssapi-disable-pafxfast=false", + expected: security.SASL{ + SASLMechanism: security.GSSAPIMechanism, + GSSAPI: security.GSSAPI{ + AuthType: security.UserAuth, + KerberosConfigPath: "/root/config", + ServiceName: "a", + Username: "user", + Password: "pwd", + Realm: "realm", + }, + }, + }, + { + name: "valid GSSAPI keytab auth SASL", + uri: baseURI + "?sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=keytab" + + "&sasl-gssapi-kerberos-config-path=/root/config" + + "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + + "&sasl-gssapi-keytab-path=/root/keytab&sasl-gssapi-realm=realm" + + "&sasl-gssapi-disable-pafxfast=false", + expected: security.SASL{ + SASLMechanism: security.GSSAPIMechanism, + GSSAPI: security.GSSAPI{ + AuthType: security.KeyTabAuth, + KeyTabPath: "/root/keytab", + KerberosConfigPath: "/root/config", + ServiceName: "a", + Username: "user", + Realm: "realm", + }, + }, + }, + { + name: "invalid mechanism", + uri: baseURI + "?sasl-mechanism=a", + expectErr: "unknown a SASL mechanism", + }, + { + name: "invalid GSSAPI auth type", + uri: baseURI + "?sasl-mechanism=gssapi&sasl-gssapi-auth-type=keyta1b", + expectErr: "unknown keyta1b auth type", + }, + { + name: "valid OAUTHBEARER SASL", + uri: baseURI + "?sasl-mechanism=OAUTHBEARER", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientID: aws.String("client_id"), + SASLOAuthClientSecret: aws.String("Y2xpZW50X3NlY3JldA=="), + SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), + }, + expected: security.SASL{ + SASLMechanism: security.OAuthMechanism, + OAuth2: security.OAuth2{ + ClientID: "client_id", + ClientSecret: "client_secret", + TokenURL: "127.0.0.1:9093/token", + GrantType: "client_credentials", + }, + }, + }, + { + name: "invalid OAUTHBEARER SASL: missing client id", + uri: baseURI + "?sasl-mechanism=OAUTHBEARER", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientSecret: aws.String("Y2xpZW50X3NlY3JldA=="), + SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), + }, + expectErr: "OAuth2 client id is empty", + }, + { + name: "invalid OAUTHBEARER SASL: missing client secret", + uri: baseURI + "?sasl-mechanism=OAUTHBEARER", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientID: aws.String("client_id"), + SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), + }, + expectErr: "OAuth2 client secret is empty", + }, + { + name: "invalid OAUTHBEARER SASL: missing token url", + uri: baseURI + "?sasl-mechanism=OAUTHBEARER", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientID: aws.String("client_id"), + SASLOAuthClientSecret: aws.String("Y2xpZW50X3NlY3JldA=="), + }, + expectErr: "OAuth2 token url is empty", + }, + { + name: "invalid OAUTHBEARER SASL: non base64 client secret", + uri: baseURI + "?sasl-mechanism=OAUTHBEARER", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientID: aws.String("client_id"), + SASLOAuthClientSecret: aws.String("client_secret"), + SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), + }, + expectErr: "OAuth2 client secret is not base64 encoded", + }, + { + name: "invalid OAUTHBEARER SASL: wrong mechanism", + uri: baseURI + "?sasl-mechanism=GSSAPI", + kafkaConfig: &config.KafkaConfig{ + SASLOAuthClientID: aws.String("client_id"), + SASLOAuthClientSecret: aws.String("Y2xpZW50X3NlY3JldA=="), + SASLOAuthTokenURL: aws.String("127.0.0.1:9093/token"), + }, + expectErr: "OAuth2 is only supported with SASL mechanism type OAUTHBEARER", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + sinkURI, err := url.Parse(test.uri) + require.NoError(t, err) + replicaConfig := config.GetDefaultReplicaConfig() + replicaConfig.Sink.KafkaConfig = test.kafkaConfig + options := NewOptions() + err = options.Apply( + common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), + sinkURI, + replicaConfig.Sink, + ) + if test.expectErr != "" { + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + require.ErrorContains(t, err, test.expectErr) + return + } + + require.NoError(t, err) + require.Equal(t, test.expected, *options.SASL) + }) + } +} + +func TestApplyTLS(t *testing.T) { + t.Parallel() + + const baseURI = "kafka://127.0.0.1:9092/abc" + tests := []struct { + name string + uri string + expectedTLS bool + expectedCredential security.Credential + expectErr string + }{ + { + name: "tls config with enable-tls set to true", + uri: baseURI + "?enable-tls=true", + expectedTLS: true, + }, + { + name: "tls config with no enable-tls and credential files supplied", + uri: baseURI + "?ca=/root/ca.file&cert=/root/cert.file&key=/root/key.file", + expectedCredential: security.Credential{ + CAPath: "/root/ca.file", + CertPath: "/root/cert.file", + KeyPath: "/root/key.file", + }, + expectedTLS: true, + }, + {name: "tls config with no enable-tls and no credential files", uri: baseURI}, + { + name: "tls config with enable-tls false and credential files supplied", + uri: baseURI + "?enable-tls=false&ca=/root/ca&cert=/root/cert&key=/root/key", + expectedCredential: security.Credential{ + CAPath: "/root/ca", + CertPath: "/root/cert", + KeyPath: "/root/key", + }, + expectErr: "credential files are supplied, but 'enable-tls' is set to false", + }, + { + name: "tls config with incomplete credential files", + uri: baseURI + "?enable-tls=true&ca=/root/ca&cert=/root/cert", + expectedCredential: security.Credential{ + CAPath: "/root/ca", + CertPath: "/root/cert", + }, + expectErr: "ca, cert and key files should all be supplied", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + sinkURI, err := url.Parse(test.uri) + require.NoError(t, err) + options := NewOptions() + err = options.Apply( + common.NewChangefeedID4Test(common.DefaultKeyspaceName, "test"), + sinkURI, + config.GetDefaultReplicaConfig().Sink, + ) + if test.expectErr != "" { + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + require.ErrorContains(t, err, test.expectErr) + } else { + require.NoError(t, err) + } + require.Equal(t, test.expectedTLS, options.EnableTLS) + require.Equal(t, test.expectedCredential, *options.Credential) + }) + } +} + func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) { tests := []struct { name string diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 65ebb088ad..0aeaf67ead 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -15,12 +15,10 @@ package kafka import ( "context" - "net/http" "net/url" "testing" "github.com/IBM/sarama" - "github.com/gin-gonic/gin/binding" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" @@ -212,271 +210,6 @@ func TestNewSaramaConfigMaxRetryFromSinkURI(t *testing.T) { } } -func TestApplySASL(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - URI string - replicaConfig func() *config.ReplicaConfig - exceptErr string - }{ - { - name: "no params", - URI: "kafka://127.0.0.1:9092/abc", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "", - }, - { - name: "valid PLAIN SASL", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "", - }, - { - name: "valid SCRAM SASL", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=SCRAM-SHA-512", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "", - }, - { - name: "valid GSSAPI user auth SASL", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=USER" + - "&sasl-gssapi-kerberos-config-path=/root/config" + - "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + - "&sasl-gssapi-password=pwd" + - "&sasl-gssapi-realm=realm&sasl-gssapi-disable-pafxfast=false", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "", - }, - { - name: "valid GSSAPI keytab auth SASL", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-mechanism=GSSAPI&sasl-gssapi-auth-type=keytab" + - "&sasl-gssapi-kerberos-config-path=/root/config" + - "&sasl-gssapi-service-name=a&sasl-gssapi-user=user" + - "&sasl-gssapi-keytab-path=/root/keytab" + - "&sasl-gssapi-realm=realm&sasl-gssapi-disable-pafxfast=false", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "", - }, - { - name: "invalid mechanism", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-mechanism=a", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "unknown a SASL mechanism", - }, - { - name: "invalid GSSAPI auth type", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-mechanism=gssapi&sasl-gssapi-auth-type=keyta1b", - replicaConfig: config.GetDefaultReplicaConfig, - exceptErr: "unknown keyta1b auth type", - }, - { - name: "valid OAUTHBEARER SASL", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=OAUTHBEARER", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientID := "client_id" - clientSecret := "Y2xpZW50X3NlY3JldA==" // base64(client_secret) - tokenURL := "127.0.0.1:9093/token" - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientID: &clientID, - SASLOAuthClientSecret: &clientSecret, - SASLOAuthTokenURL: &tokenURL, - } - return cfg - }, - exceptErr: "", - }, - { - name: "invalid OAUTHBEARER SASL: missing client id", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=OAUTHBEARER", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientSecret := "Y2xpZW50X3NlY3JldA==" // base64(client_secret) - tokenURL := "127.0.0.1:9093/token" - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientSecret: &clientSecret, - SASLOAuthTokenURL: &tokenURL, - } - return cfg - }, - exceptErr: "OAuth2 client id is empty", - }, - { - name: "invalid OAUTHBEARER SASL: missing client secret", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=OAUTHBEARER", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientID := "client_id" - tokenURL := "127.0.0.1:9093/token" - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientID: &clientID, - SASLOAuthTokenURL: &tokenURL, - } - return cfg - }, - exceptErr: "OAuth2 client secret is empty", - }, - { - name: "invalid OAUTHBEARER SASL: missing token url", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=OAUTHBEARER", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientID := "client_id" - clientSecret := "Y2xpZW50X3NlY3JldA==" // base64(client_secret) - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientID: &clientID, - SASLOAuthClientSecret: &clientSecret, - } - return cfg - }, - exceptErr: "OAuth2 token url is empty", - }, - { - name: "invalid OAUTHBEARER SASL: non base64 client secret", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=OAUTHBEARER", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientID := "client_id" - clientSecret := "client_secret" - tokenURL := "127.0.0.1:9093/token" - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientID: &clientID, - SASLOAuthClientSecret: &clientSecret, - SASLOAuthTokenURL: &tokenURL, - } - return cfg - }, - exceptErr: "OAuth2 client secret is not base64 encoded", - }, - { - name: "invalid OAUTHBEARER SASL: wrong mechanism", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0&sasl-mechanism=GSSAPI", - replicaConfig: func() *config.ReplicaConfig { - cfg := config.GetDefaultReplicaConfig() - oauthMechanism := string(security.OAuthMechanism) - clientID := "client_id" - clientSecret := "Y2xpZW50X3NlY3JldA==" // base64(client_secret) - tokenURL := "127.0.0.1:9093/token" - cfg.Sink.KafkaConfig = &config.KafkaConfig{ - SASLMechanism: &oauthMechanism, - SASLOAuthClientID: &clientID, - SASLOAuthClientSecret: &clientSecret, - SASLOAuthTokenURL: &tokenURL, - } - return cfg - }, - exceptErr: "OAuth2 is only supported with SASL mechanism type OAUTHBEARER", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - t.Parallel() - options := NewOptions() - sinkURI, err := url.Parse(test.URI) - require.NoError(t, err) - req := &http.Request{URL: sinkURI} - urlParameter := &urlConfig{} - err = binding.Query.Bind(req, urlParameter) - require.NoError(t, err) - if test.exceptErr == "" { - require.Nil(t, options.applySASL(urlParameter, test.replicaConfig().Sink)) - } else { - require.Regexp(t, test.exceptErr, - options.applySASL(urlParameter, test.replicaConfig().Sink).Error()) - } - }) - } -} - -func TestApplyTLS(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - URI string - tlsEnabled bool - exceptErr string - }{ - { - name: "tls config with 'enable-tls' set to true", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain&enable-tls=true", - tlsEnabled: true, - exceptErr: "", - }, - { - name: "tls config with no 'enable-tls', and credential files are supplied", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain" + - "&ca=/root/ca.file&cert=/root/cert.file&key=/root/key.file", - tlsEnabled: true, - exceptErr: "", - }, - { - name: "tls config with no 'enable-tls', and credential files are not supplied", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain", - tlsEnabled: false, - exceptErr: "", - }, - { - name: "tls config with 'enable-tls' set to false, and credential files are supplied", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain&enable-tls=false" + - "&ca=/root/ca&cert=/root/cert&key=/root/key", - tlsEnabled: false, - exceptErr: "credential files are supplied, but 'enable-tls' is set to false", - }, - { - name: "tls config with 'enable-tls' set to true, and some of " + - "the credential files are not supplied ", - URI: "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&partition-num=0" + - "&sasl-user=user&sasl-password=password&sasl-mechanism=plain&enable-tls=true" + - "&ca=/root/ca&cert=/root/cert&", - tlsEnabled: false, - exceptErr: "ca, cert and key files should all be supplied", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - t.Parallel() - options := NewOptions() - sinkURI, err := url.Parse(test.URI) - require.NoError(t, err) - req := &http.Request{URL: sinkURI} - urlParameter := &urlConfig{} - err = binding.Query.Bind(req, urlParameter) - require.NoError(t, err) - if test.exceptErr == "" { - require.Nil(t, options.applyTLS(urlParameter)) - } else { - require.Regexp(t, test.exceptErr, options.applyTLS(urlParameter).Error()) - } - require.Equal(t, test.tlsEnabled, options.EnableTLS) - }) - } -} - func TestCompleteSaramaSASLConfig(t *testing.T) { t.Parallel() From 536d647351dab4fb27d38c34c962bb07f196bb18 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 18:31:35 +0800 Subject: [PATCH 11/13] remove coupled oauth2 test --- pkg/sink/kafka/sarama_config_test.go | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 0aeaf67ead..051c0f59f2 100644 --- a/pkg/sink/kafka/sarama_config_test.go +++ b/pkg/sink/kafka/sarama_config_test.go @@ -141,21 +141,6 @@ func TestSelectKafkaVersion(t *testing.T) { } } -func TestNewSaramaConfigInvalidOAuthTokenURL(t *testing.T) { - options := NewOptions() - options.SASL = &security.SASL{ - SASLMechanism: security.OAuthMechanism, - OAuth2: security.OAuth2{ - TokenURL: "http://test.com/Segment%%2815197306101420000%29", - }, - } - - _, err := newSaramaConfig(t.Context(), options) - require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) - var escapeErr url.EscapeError - require.ErrorAs(t, err, &escapeErr) -} - func TestNewSaramaConfigMaxRetryFromSinkURI(t *testing.T) { t.Parallel() From 97f1ec5af15d8c11f46b024b2d65ab77e998dc0a Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 18:40:16 +0800 Subject: [PATCH 12/13] fix oauth2 token provider test --- pkg/sink/kafka/oauth2_token_provider_test.go | 148 +++++++++++++------ 1 file changed, 102 insertions(+), 46 deletions(-) diff --git a/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/oauth2_token_provider_test.go index 0ed4d7c044..f6eda43b74 100644 --- a/pkg/sink/kafka/oauth2_token_provider_test.go +++ b/pkg/sink/kafka/oauth2_token_provider_test.go @@ -14,65 +14,121 @@ package kafka import ( - "context" + "io" + "net/http" + "net/http/httptest" "net/url" "testing" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/security" "github.com/stretchr/testify/require" + "golang.org/x/oauth2" ) -func TestNewTokenProvider(t *testing.T) { +func TestNewTokenProviderRejectsInvalidTokenURL(t *testing.T) { t.Parallel() - for _, test := range []struct { - name string - options *options - expectedErr string - }{ - { - name: "valid", - options: &options{ - SASL: &security.SASL{ - OAuth2: security.OAuth2{ - ClientID: "client-id", - ClientSecret: "client-secret", - TokenURL: "http://localhost:8080/oauth2/token", - Scopes: []string{"scope1", "scope2"}, - GrantType: "client_credentials", - }, - }, + options := &options{ + SASL: &security.SASL{ + OAuth2: security.OAuth2{ + ClientID: "client-id", + ClientSecret: "client-secret", + TokenURL: "http://test.com/Segment%%2815197306101420000%29", + Scopes: []string{"scope1", "scope2"}, + GrantType: "client_credentials", }, }, - { - name: "invalid token URL", - options: &options{ - SASL: &security.SASL{ - OAuth2: security.OAuth2{ - ClientID: "client-id", - ClientSecret: "client-secret", - TokenURL: "http://test.com/Segment%%2815197306101420000%29", - Scopes: []string{"scope1", "scope2"}, - GrantType: "client_credentials", - }, - }, + } + + _, err := newTokenProvider(t.Context(), options) + require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) + var escapeErr url.EscapeError + require.ErrorAs(t, err, &escapeErr) + require.ErrorContains(t, err, "invalid URL escape") +} + +func TestTokenProviderRequestsToken(t *testing.T) { + t.Parallel() + + type tokenRequest struct { + method string + path string + form url.Values + err error + } + requestCh := make(chan tokenRequest, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + err := r.ParseForm() + requestCh <- tokenRequest{ + method: r.Method, + path: r.URL.Path, + form: r.PostForm, + err: err, + } + + w.Header().Set("Content-Type", "application/json") + if _, err := io.WriteString(w, `{"access_token":"access-token","token_type":"bearer"}`); err != nil { + t.Errorf("write token response: %v", err) + } + })) + t.Cleanup(server.Close) + + options := &options{ + SASL: &security.SASL{ + OAuth2: security.OAuth2{ + ClientID: "client-id", + ClientSecret: "client-secret", + TokenURL: server.URL + "/oauth2/token", + Scopes: []string{"scope1", "scope2"}, + GrantType: "custom_grant", + Audience: "test-audience", + }, + }, + } + + provider, err := newTokenProvider(t.Context(), options) + require.NoError(t, err) + _, err = provider.Token() + require.NoError(t, err) + + request := <-requestCh + require.NoError(t, request.err) + require.Equal(t, http.MethodPost, request.method) + require.Equal(t, "/oauth2/token", request.path) + require.Equal(t, "custom_grant", request.form.Get("grant_type")) + require.Equal(t, "test-audience", request.form.Get("audience")) + require.Equal(t, "scope1 scope2", request.form.Get("scope")) +} + +func TestTokenProviderPropagatesEndpointError(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusUnauthorized) + if _, err := io.WriteString(w, `{"error":"invalid_client","error_description":"bad credentials"}`); err != nil { + t.Errorf("write token error response: %v", err) + } + })) + t.Cleanup(server.Close) + + options := &options{ + SASL: &security.SASL{ + OAuth2: security.OAuth2{ + ClientID: "client-id", + ClientSecret: "client-secret", + TokenURL: server.URL, }, - expectedErr: "invalid URL escape", }, - } { - ts := test - t.Run(ts.name, func(t *testing.T) { - t.Parallel() - _, err := newTokenProvider(context.TODO(), ts.options) - if ts.expectedErr == "" { - require.NoError(t, err) - } else { - require.ErrorIs(t, err, errors.ErrKafkaInvalidConfig) - var escapeErr url.EscapeError - require.ErrorAs(t, err, &escapeErr) - require.Contains(t, err.Error(), ts.expectedErr) - } - }) } + + provider, err := newTokenProvider(t.Context(), options) + require.NoError(t, err) + _, err = provider.Token() + var retrieveErr *oauth2.RetrieveError + require.ErrorAs(t, err, &retrieveErr) + require.Equal(t, http.StatusUnauthorized, retrieveErr.Response.StatusCode) + require.Equal(t, "invalid_client", retrieveErr.ErrorCode) + require.Equal(t, "bad credentials", retrieveErr.ErrorDescription) } From c07f9ca40632caa5ecca05e571f9caaa0d382e82 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 6 Aug 2026 10:54:29 +0800 Subject: [PATCH 13/13] fix tests --- pkg/sink/kafka/oauth2_token_provider_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/oauth2_token_provider_test.go index f6eda43b74..d796d80d4e 100644 --- a/pkg/sink/kafka/oauth2_token_provider_test.go +++ b/pkg/sink/kafka/oauth2_token_provider_test.go @@ -89,8 +89,9 @@ func TestTokenProviderRequestsToken(t *testing.T) { provider, err := newTokenProvider(t.Context(), options) require.NoError(t, err) - _, err = provider.Token() + token, err := provider.Token() require.NoError(t, err) + require.Equal(t, "access-token", token.Token) request := <-requestCh require.NoError(t, request.err)