kafka(ticdc): sarama admin client fetch metadata by cache (#9511)

cdc/sink/mq/manager/kafka_manager.go

@@ -61,7 +61,7 @@ func NewKafkaTopicManager(
 	ctx context.Context,
 	admin kafka.ClusterAdminClient,
 	cfg *kafkaconfig.AutoCreateTopicConfig,
-) (*kafkaTopicManager, error) {
+) *kafkaTopicManager {
 	changefeedID := contextutil.ChangefeedIDFromCtx(ctx)
 	mgr := &kafkaTopicManager{
 		changefeedID:      changefeedID,
@@ -74,7 +74,7 @@ func NewKafkaTopicManager(
 	// Background refresh metadata.
 	go mgr.backgroundRefreshMeta(ctx)
 
-	return mgr, nil
+	return mgr
 }
 
 // GetPartitionNum returns the number of partitions of the topic.

cdc/sink/mq/manager/kafka_manager_test.go

@@ -35,11 +35,9 @@ func TestPartitions(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager, err := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
-	require.Nil(t, err)
+	manager := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
 	defer manager.Close()
-	partitionsNum, err := manager.GetPartitionNum(
-		kafka.DefaultMockTopicName)
+	partitionsNum, err := manager.GetPartitionNum(kafka.DefaultMockTopicName)
 	require.Nil(t, err)
 	require.Equal(t, int32(3), partitionsNum)
 }
@@ -59,8 +57,7 @@ func TestCreateTopic(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager, err := NewKafkaTopicManager(ctx, adminClient, cfg)
-	require.Nil(t, err)
+	manager := NewKafkaTopicManager(ctx, adminClient, cfg)
 	defer manager.Close()
 	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(kafka.DefaultMockTopicName)
 	require.Nil(t, err)
@@ -75,8 +72,7 @@ func TestCreateTopic(t *testing.T) {
 
 	// Try to create a topic without auto create.
 	cfg.AutoCreate = false
-	manager, err = NewKafkaTopicManager(ctx, adminClient, cfg)
-	require.Nil(t, err)
+	manager = NewKafkaTopicManager(ctx, adminClient, cfg)
 	defer manager.Close()
 	_, err = manager.CreateTopicAndWaitUntilVisible("new-topic2")
 	require.Regexp(
@@ -92,8 +88,7 @@ func TestCreateTopic(t *testing.T) {
 		PartitionNum:      2,
 		ReplicationFactor: 4,
 	}
-	manager, err = NewKafkaTopicManager(ctx, adminClient, cfg)
-	require.Nil(t, err)
+	manager = NewKafkaTopicManager(ctx, adminClient, cfg)
 	defer manager.Close()
 	_, err = manager.CreateTopicAndWaitUntilVisible("new-topic-failed")
 	require.Regexp(
@@ -116,8 +111,7 @@ func TestCreateTopicWithDelay(t *testing.T) {
 		ReplicationFactor: 1,
 	}
 
-	manager, err := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
-	require.Nil(t, err)
+	manager := NewKafkaTopicManager(context.TODO(), adminClient, cfg)
 	defer manager.Close()
 	partitionNum, err := manager.createTopic("new_topic")
 	require.Nil(t, err)

cdc/sink/mq/mq.go

@@ -432,15 +432,7 @@ func NewKafkaSaramaSink(ctx context.Context, sinkURI *url.URL,
 		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
 	}
 
-	topicManager, err := manager.NewKafkaTopicManager(
-		ctx,
-		adminClient,
-		baseConfig.DeriveTopicConfig(),
-	)
-	if err != nil {
-		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
-	}
-
+	topicManager := manager.NewKafkaTopicManager(ctx, adminClient, baseConfig.DeriveTopicConfig())
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(topic); err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err)
 	}

cdc/sink/mq/producer/kafka/config.go

@@ -434,21 +434,21 @@ func NewSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	}
 	config.Version = version
 
-	// Producer fetch metadata from brokers frequently, if metadata cannot be
-	// refreshed easily, this would indicate the network condition between the
+	// Admin client would refresh metadata periodically,
+	// if metadata cannot be refreshed easily, this would indicate the network condition between the
 	// capture server and kafka broker is not good.
-	// In the scenario that cannot get response from Kafka server, this default
-	// setting can help to get response more quickly.
-	config.Metadata.Retry.Max = 1
-	config.Metadata.Retry.Backoff = 100 * time.Millisecond
+	// Set the timeout to 2 minutes to ensure that the underlying client does not retry for too long.
+	// If retrying to obtain the metadata fails, simply return the error and let sinkManager rebuild the sink.
+	config.Metadata.Retry.Max = 10
+	config.Metadata.Retry.Backoff = 200 * time.Millisecond
 	// This Timeout is useless if the `RefreshMetadata` time cost is less than it.
-	config.Metadata.Timeout = 1 * time.Minute
+	config.Metadata.Timeout = 2 * time.Minute
 
-	// Admin.Retry take effect on `ClusterAdmin` related operations,
-	// only `CreateTopic` for cdc now. set the `Timeout` to `1m` to make CI stable.
-	config.Admin.Retry.Max = 5
-	config.Admin.Retry.Backoff = 100 * time.Millisecond
-	config.Admin.Timeout = 1 * time.Minute
+	config.Admin.Retry.Max = 10
+	config.Admin.Retry.Backoff = 200 * time.Millisecond
+	// This timeout control the request timeout for each admin request.
+	// set it as the read timeout.
+	config.Admin.Timeout = 10 * time.Second
 
 	// Producer.Retry take effect when the producer try to send message to kafka
 	// brokers. If kafka cluster is healthy, just the default value should be enough.

cdc/sinkv2/util/helper.go

@@ -94,15 +94,7 @@ func GetTopicManagerAndTryCreateTopic(
 	topicCfg *kafka.AutoCreateTopicConfig,
 	adminClient pkafka.ClusterAdminClient,
 ) (manager.TopicManager, error) {
-	topicManager, err := manager.NewKafkaTopicManager(
-		ctx,
-		adminClient,
-		topicCfg,
-	)
-	if err != nil {
-		return nil, cerror.WrapError(cerror.ErrKafkaNewSaramaProducer, err)
-	}
-
+	topicManager := manager.NewKafkaTopicManager(ctx, adminClient, topicCfg)
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(topic); err != nil {
 		return nil, cerror.WrapError(cerror.ErrKafkaCreateTopic, err)
 	}