kafka: decouple sarama from tests to make it library independent

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
 }

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
 	}

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 {

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)
 

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,

pkg/sink/kafka/cluster_admin_client.go → 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