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.go b/downstreamadapter/sink/kafka/sink.go index 6c6332a4ae..caf9f324cb 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -72,6 +72,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 { @@ -112,7 +116,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 8b205c1649..f4e7860d4c 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" @@ -41,6 +40,10 @@ import ( const kafkaSinkTestTopic = "mock_topic" +type noopMetricsCollector struct{} + +func (noopMetricsCollector) Run(context.Context) {} + func TestSinkWorkersReturnContextError(t *testing.T) { contexts := []struct { name string @@ -88,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) })) @@ -115,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}, true).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") @@ -158,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: { @@ -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/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index e436e8801a..8e43a52df6 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 @@ -52,7 +52,7 @@ type kafkaTopicManager struct { func newKafkaTopicManager( defaultTopic string, changefeedID common.ChangeFeedID, - admin kafka.ClusterAdminClient, + admin kafka.AdminClient, cfg *kafka.AutoCreateTopicConfig, ) *kafkaTopicManager { return &kafkaTopicManager{ @@ -69,7 +69,7 @@ func EnsureTopic( changefeedID common.ChangeFeedID, topic string, topicCfg *kafka.AutoCreateTopicConfig, - adminClient kafka.ClusterAdminClient, + adminClient kafka.AdminClient, ) error { topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg) _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic) @@ -82,7 +82,7 @@ func GetTopicManagerAndTryCreateTopic( changefeedID common.ChangeFeedID, topic string, topicCfg *kafka.AutoCreateTopicConfig, - adminClient kafka.ClusterAdminClient, + adminClient kafka.AdminClient, ) (TopicManager, error) { topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index b00bf947d0..9e4b7e32b0 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,12 @@ func (m *mockAdminClientWithDeniedDescribe) GetTopicsMeta( if ignoreTopicError { return map[string]kafka.TopicDetail{}, nil } - return nil, sarama.ErrTopicAuthorizationFailed + return nil, errors.WrapError( + errors.ErrKafkaAdminAPI, + sarama.ErrTopicAuthorizationFailed, + "describe-topic", + topics[0], + ) } func (m *mockAdminClientWithDeniedDescribe) CreateTopic( @@ -52,7 +57,7 @@ func (m *mockAdminClientWithDeniedDescribe) CreateTopic( } type mockAdminClientWithDeniedCreate struct { - *kafka.MockClusterAdminClient + *kafka.MockAdminClient createTopicCalled bool describeCount int } @@ -69,14 +74,19 @@ func (m *mockAdminClientWithDeniedCreate) CreateTopic( detail *kafka.TopicDetail, ) error { m.createTopicCalled = true - return sarama.ErrClusterAuthorizationFailed + return errors.WrapError( + errors.ErrKafkaAdminAPI, + sarama.ErrClusterAuthorizationFailed, + "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, @@ -180,7 +190,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). @@ -211,7 +221,7 @@ func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, PartitionNum: 2, @@ -254,7 +264,7 @@ func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) { t.Parallel() ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) topic := "existing-topic" adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).Return( map[string]kafka.TopicDetail{ @@ -282,7 +292,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, @@ -311,7 +321,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/sink/kafka/cluster_admin_client.go b/pkg/sink/kafka/admin_client.go similarity index 94% rename from pkg/sink/kafka/cluster_admin_client.go rename to pkg/sink/kafka/admin_client.go index d378658c3a..cea348c2a2 100644 --- a/pkg/sink/kafka/cluster_admin_client.go +++ b/pkg/sink/kafka/admin_client.go @@ -25,9 +25,9 @@ type Broker struct { ID int32 } -// ClusterAdminClient is the administrative client for Kafka, +// AdminClient is the administrative client for Kafka, // which supports managing and inspecting topics, brokers, configurations and ACLs. -type ClusterAdminClient interface { +type AdminClient interface { // GetAllBrokers return all brokers among the cluster GetAllBrokers() []Broker diff --git a/pkg/sink/kafka/cluster_admin_client_mock.go b/pkg/sink/kafka/admin_client_mock.go similarity index 51% rename from pkg/sink/kafka/cluster_admin_client_mock.go rename to pkg/sink/kafka/admin_client_mock.go index 4fb06a2a56..2d2778ea3b 100644 --- a/pkg/sink/kafka/cluster_admin_client_mock.go +++ b/pkg/sink/kafka/admin_client_mock.go @@ -1,5 +1,5 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: pkg/sink/kafka/cluster_admin_client.go +// Source: pkg/sink/kafka/admin_client.go // Package kafka is a generated GoMock package. package kafka @@ -10,43 +10,43 @@ import ( gomock "github.com/golang/mock/gomock" ) -// MockClusterAdminClient is a mock of ClusterAdminClient interface. -type MockClusterAdminClient struct { +// MockAdminClient is a mock of AdminClient interface. +type MockAdminClient struct { ctrl *gomock.Controller - recorder *MockClusterAdminClientMockRecorder + recorder *MockAdminClientMockRecorder } -// MockClusterAdminClientMockRecorder is the mock recorder for MockClusterAdminClient. -type MockClusterAdminClientMockRecorder struct { - mock *MockClusterAdminClient +// MockAdminClientMockRecorder is the mock recorder for MockAdminClient. +type MockAdminClientMockRecorder struct { + mock *MockAdminClient } -// NewMockClusterAdminClient creates a new mock instance. -func NewMockClusterAdminClient(ctrl *gomock.Controller) *MockClusterAdminClient { - mock := &MockClusterAdminClient{ctrl: ctrl} - mock.recorder = &MockClusterAdminClientMockRecorder{mock} +// 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 *MockClusterAdminClient) EXPECT() *MockClusterAdminClientMockRecorder { +func (m *MockAdminClient) EXPECT() *MockAdminClientMockRecorder { return m.recorder } // Close mocks base method. -func (m *MockClusterAdminClient) Close() { +func (m *MockAdminClient) Close() { m.ctrl.T.Helper() m.ctrl.Call(m, "Close") } // Close indicates an expected call of Close. -func (mr *MockClusterAdminClientMockRecorder) Close() *gomock.Call { +func (mr *MockAdminClientMockRecorder) Close() *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockClusterAdminClient)(nil).Close)) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockAdminClient)(nil).Close)) } // CreateTopic mocks base method. -func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail) error { +func (m *MockAdminClient) CreateTopic(detail *TopicDetail) error { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "CreateTopic", detail) ret0, _ := ret[0].(error) @@ -54,13 +54,13 @@ func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail) error { } // CreateTopic indicates an expected call of CreateTopic. -func (mr *MockClusterAdminClientMockRecorder) CreateTopic(detail interface{}) *gomock.Call { +func (mr *MockAdminClientMockRecorder) CreateTopic(detail interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockClusterAdminClient)(nil).CreateTopic), detail) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockAdminClient)(nil).CreateTopic), detail) } // GetAllBrokers mocks base method. -func (m *MockClusterAdminClient) GetAllBrokers() []Broker { +func (m *MockAdminClient) GetAllBrokers() []Broker { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "GetAllBrokers") ret0, _ := ret[0].([]Broker) @@ -68,13 +68,13 @@ func (m *MockClusterAdminClient) GetAllBrokers() []Broker { } // GetAllBrokers indicates an expected call of GetAllBrokers. -func (mr *MockClusterAdminClientMockRecorder) GetAllBrokers() *gomock.Call { +func (mr *MockAdminClientMockRecorder) GetAllBrokers() *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllBrokers", reflect.TypeOf((*MockClusterAdminClient)(nil).GetAllBrokers)) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllBrokers", reflect.TypeOf((*MockAdminClient)(nil).GetAllBrokers)) } // GetBrokerConfig mocks base method. -func (m *MockClusterAdminClient) GetBrokerConfig(configName string) (string, bool, error) { +func (m *MockAdminClient) GetBrokerConfig(configName string) (string, bool, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "GetBrokerConfig", configName) ret0, _ := ret[0].(string) @@ -84,13 +84,13 @@ func (m *MockClusterAdminClient) GetBrokerConfig(configName string) (string, boo } // GetBrokerConfig indicates an expected call of GetBrokerConfig. -func (mr *MockClusterAdminClientMockRecorder) GetBrokerConfig(configName interface{}) *gomock.Call { +func (mr *MockAdminClientMockRecorder) GetBrokerConfig(configName interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBrokerConfig", reflect.TypeOf((*MockClusterAdminClient)(nil).GetBrokerConfig), configName) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetBrokerConfig", reflect.TypeOf((*MockAdminClient)(nil).GetBrokerConfig), configName) } // GetTopicConfig mocks base method. -func (m *MockClusterAdminClient) GetTopicConfig(topicName, configName string) (string, bool, error) { +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) @@ -100,13 +100,13 @@ func (m *MockClusterAdminClient) GetTopicConfig(topicName, configName string) (s } // GetTopicConfig indicates an expected call of GetTopicConfig. -func (mr *MockClusterAdminClientMockRecorder) GetTopicConfig(topicName, configName interface{}) *gomock.Call { +func (mr *MockAdminClientMockRecorder) 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) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicConfig", reflect.TypeOf((*MockAdminClient)(nil).GetTopicConfig), topicName, configName) } // GetTopicsMeta mocks base method. -func (m *MockClusterAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool) (map[string]TopicDetail, error) { +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) @@ -115,13 +115,13 @@ func (m *MockClusterAdminClient) GetTopicsMeta(topics []string, ignoreTopicError } // GetTopicsMeta indicates an expected call of GetTopicsMeta. -func (mr *MockClusterAdminClientMockRecorder) GetTopicsMeta(topics, ignoreTopicError interface{}) *gomock.Call { +func (mr *MockAdminClientMockRecorder) 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) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsMeta", reflect.TypeOf((*MockAdminClient)(nil).GetTopicsMeta), topics, ignoreTopicError) } // GetTopicsPartitionsNum mocks base method. -func (m *MockClusterAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) { +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) @@ -130,7 +130,7 @@ func (m *MockClusterAdminClient) GetTopicsPartitionsNum(topics []string) (map[st } // GetTopicsPartitionsNum indicates an expected call of GetTopicsPartitionsNum. -func (mr *MockClusterAdminClientMockRecorder) GetTopicsPartitionsNum(topics interface{}) *gomock.Call { +func (mr *MockAdminClientMockRecorder) GetTopicsPartitionsNum(topics interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsPartitionsNum", reflect.TypeOf((*MockClusterAdminClient)(nil).GetTopicsPartitionsNum), topics) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopicsPartitionsNum", reflect.TypeOf((*MockAdminClient)(nil).GetTopicsPartitionsNum), topics) } 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/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/pkg/sink/kafka/oauth2_token_provider_test.go b/pkg/sink/kafka/oauth2_token_provider_test.go index 0ed4d7c044..d796d80d4e 100644 --- a/pkg/sink/kafka/oauth2_token_provider_test.go +++ b/pkg/sink/kafka/oauth2_token_provider_test.go @@ -14,65 +14,122 @@ 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) + token, err := provider.Token() + require.NoError(t, err) + require.Equal(t, "access-token", token.Token) + + 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) } 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 93498b384d..cd5f52c73e 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,12 +21,12 @@ 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" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/security" "github.com/stretchr/testify/require" ) @@ -35,134 +34,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 *MockClusterAdminClient - 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: NewMockClusterAdminClient(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()). - 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) 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() @@ -290,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 @@ -451,27 +569,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 +594,53 @@ 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) - require.NoError(t, err) - - configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture) + 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), + ) 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 +662,6 @@ func TestValidateReplicationFactor(t *testing.T) { err = localAcksConfig.ValidateReplicationFactor(adminClient) require.NoError(t, err) - adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName) missingBrokerConfig := &AutoCreateTopicConfig{ AutoCreate: true, ReplicationFactor: 1, @@ -555,15 +669,55 @@ func TestValidateReplicationFactor(t *testing.T) { } err = missingBrokerConfig.ValidateReplicationFactor(adminClient) 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) + 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) { @@ -716,39 +870,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() }) } } 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_config_test.go b/pkg/sink/kafka/sarama_config_test.go index 65ebb088ad..051c0f59f2 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" @@ -143,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() @@ -212,271 +195,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() 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, diff --git a/pkg/sink/kafka/sarama_sync_producer_test.go b/pkg/sink/kafka/sarama_sync_producer_test.go index 522dc13bcb..e00e0dafa8 100644 --- a/pkg/sink/kafka/sarama_sync_producer_test.go +++ b/pkg/sink/kafka/sarama_sync_producer_test.go @@ -19,7 +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" @@ -127,12 +126,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 c825e85946..e4f0a36b51 100755 --- a/scripts/generate-mock.sh +++ b/scripts/generate-mock.sh @@ -34,9 +34,8 @@ 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/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