kafka: create and wait for topic during verification (#5898)

downstreamadapter/sink/kafka/sink.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/topicmanager"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
@@ -122,29 +123,10 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.URL,
 	}
 	defer adminClient.Close()
 
-	topics, err := adminClient.GetTopicsMeta([]string{topic}, false)
+	err = topicmanager.EnsureTopic(ctx, changefeedID, topic, options.DeriveTopicConfig(), adminClient)
 	if err != nil {
 		return err
 	}
-	if _, exists := topics[topic]; !exists {
-		topicConfig := options.DeriveTopicConfig()
-		if !topicConfig.AutoCreate {
-			return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic)
-		}
-		if err = topicConfig.ValidateReplicationFactor(adminClient); err != nil {
-			return err
-		}
-
-		// the topic is not created, only validate.
-		err = adminClient.CreateTopic(&kafka.TopicDetail{
-			Name:              topic,
-			NumPartitions:     topicConfig.PartitionNum,
-			ReplicationFactor: topicConfig.ReplicationFactor,
-		}, true)
-		if err != nil {
-			return err
-		}
-	}
 
 	_, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck)
 	if err != nil {

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -48,6 +48,34 @@ type kafkaTopicManager struct {
 	cancel context.CancelFunc
 }
 
+// newKafkaTopicManager creates a topic manager without starting background work.
+func newKafkaTopicManager(
+	defaultTopic string,
+	changefeedID common.ChangeFeedID,
+	admin kafka.ClusterAdminClient,
+	cfg *kafka.AutoCreateTopicConfig,
+) *kafkaTopicManager {
+	return &kafkaTopicManager{
+		defaultTopic: defaultTopic,
+		changefeedID: changefeedID,
+		admin:        admin,
+		cfg:          cfg,
+	}
+}
+
+// EnsureTopic creates the topic if needed and waits until it is visible.
+func EnsureTopic(
+	ctx context.Context,
+	changefeedID common.ChangeFeedID,
+	topic string,
+	topicCfg *kafka.AutoCreateTopicConfig,
+	adminClient kafka.ClusterAdminClient,
+) error {
+	topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg)
+	_, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic)
+	return err
+}
+
 // GetTopicManagerAndTryCreateTopic returns the topic manager and try to create the topic.
 func GetTopicManagerAndTryCreateTopic(
 	ctx context.Context,
@@ -56,39 +84,18 @@ func GetTopicManagerAndTryCreateTopic(
 	topicCfg *kafka.AutoCreateTopicConfig,
 	adminClient kafka.ClusterAdminClient,
 ) (TopicManager, error) {
-	topicManager := newKafkaTopicManager(
-		ctx, topic, changefeedID, adminClient, topicCfg,
-	)
+	topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg)
 
 	if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil {
 		return nil, err
 	}
+	ctx, cancel := context.WithCancel(ctx)
+	topicManager.cancel = cancel
+	go topicManager.backgroundRefreshMeta(ctx)
 
 	return topicManager, nil
 }
 
-// NewKafkaTopicManager creates a new topic manager.
-func newKafkaTopicManager(
-	ctx context.Context,
-	defaultTopic string,
-	changefeedID common.ChangeFeedID,
-	admin kafka.ClusterAdminClient,
-	cfg *kafka.AutoCreateTopicConfig,
-) *kafkaTopicManager {
-	mgr := &kafkaTopicManager{
-		defaultTopic: defaultTopic,
-		changefeedID: changefeedID,
-		admin:        admin,
-		cfg:          cfg,
-	}
-
-	ctx, mgr.cancel = context.WithCancel(ctx)
-	// Background refresh metadata.
-	go mgr.backgroundRefreshMeta(ctx)
-
-	return mgr
-}
-
 // GetPartitionNum returns the number of partitions of the topic.
 // It may also try to update the topics' information maintained by manager.
 func (m *kafkaTopicManager) GetPartitionNum(
@@ -238,7 +245,7 @@ func (m *kafkaTopicManager) createTopic(
 		Name:              topicName,
 		NumPartitions:     m.cfg.PartitionNum,
 		ReplicationFactor: m.cfg.ReplicationFactor,
-	}, false)
+	})
 	if err != nil {
 		log.Error(
 			"kafka topic creation failed",
@@ -259,6 +266,9 @@ func (m *kafkaTopicManager) createTopic(
 }
 
 // CreateTopicAndWaitUntilVisible wraps createTopic and waitUntilTopicVisible together.
+// If topic creation fails due to insufficient permissions, allow the changefeed
+// to be created, the error will be returned later by other operations such as send messages.
+// The topic can be created or modified externally later to fix the error.
 func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
 	ctx context.Context, topicName string,
 ) (int32, error) {

downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

@@ -46,7 +46,6 @@ func (m *mockAdminClientWithDeniedDescribe) GetTopicsMeta(
 
 func (m *mockAdminClientWithDeniedDescribe) CreateTopic(
 	detail *kafka.TopicDetail,
-	validateOnly bool,
 ) error {
 	m.createTopicCalled = true
 	return nil
@@ -68,7 +67,6 @@ func (m *mockAdminClientWithDeniedCreate) GetTopicsMeta(
 
 func (m *mockAdminClientWithDeniedCreate) CreateTopic(
 	detail *kafka.TopicDetail,
-	validateOnly bool,
 ) error {
 	m.createTopicCalled = true
 	return sarama.ErrClusterAuthorizationFailed
@@ -89,9 +87,7 @@ func TestCreateTopic(t *testing.T) {
 	changefeedID := common.NewChangefeedID4Test("test", "test")
 	ctx := context.Background()
 	var gotNewTopicDetail *kafka.TopicDetail
-	var gotNewTopicValidateOnly bool
 	var gotFailedTopicDetail *kafka.TopicDetail
-	var gotFailedTopicValidateOnly bool
 	gomock.InOrder(
 		adminClient.EXPECT().GetTopicsMeta([]string{kafkaTopicManagerTestTopic}, true).Return(
 			map[string]kafka.TopicDetail{
@@ -104,10 +100,9 @@ func TestCreateTopic(t *testing.T) {
 			map[string]kafka.TopicDetail{}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return(
 			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().CreateTopic(gomock.Any(), false).DoAndReturn(
-			func(detail *kafka.TopicDetail, validateOnly bool) error {
+		adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn(
+			func(detail *kafka.TopicDetail) error {
 				gotNewTopicDetail = detail
-				gotNewTopicValidateOnly = validateOnly
 				return nil
 			}),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return(
@@ -125,16 +120,14 @@ func TestCreateTopic(t *testing.T) {
 			map[string]kafka.TopicDetail{}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic-failed"}, false).Return(
 			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().CreateTopic(gomock.Any(), false).DoAndReturn(
-			func(detail *kafka.TopicDetail, validateOnly bool) error {
+		adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn(
+			func(detail *kafka.TopicDetail) error {
 				gotFailedTopicDetail = detail
-				gotFailedTopicValidateOnly = validateOnly
 				return errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidReplicationFactor, "create-topic", detail.Name)
 			}),
 	)
 
-	manager := newKafkaTopicManager(ctx, kafkaTopicManagerTestTopic, changefeedID, adminClient, cfg)
-	defer manager.Close()
+	manager := newKafkaTopicManager(kafkaTopicManagerTestTopic, changefeedID, adminClient, cfg)
 	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, kafkaTopicManagerTestTopic)
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
@@ -148,7 +141,6 @@ func TestCreateTopic(t *testing.T) {
 		NumPartitions:     2,
 		ReplicationFactor: 1,
 	}, gotNewTopicDetail)
-	require.False(t, gotNewTopicValidateOnly)
 	partitionsNum, err := manager.GetPartitionNum(ctx, "new-topic")
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionsNum)
@@ -160,8 +152,7 @@ func TestCreateTopic(t *testing.T) {
 		ReplicationFactor: 1,
 		RequiredAcks:      kafka.WaitForAll,
 	}
-	manager = newKafkaTopicManager(ctx, "new-topic2", changefeedID, adminClient, cfg)
-	defer manager.Close()
+	manager = newKafkaTopicManager("new-topic2", changefeedID, adminClient, cfg)
 	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic2")
 	require.Regexp(
 		t,
@@ -177,14 +168,12 @@ func TestCreateTopic(t *testing.T) {
 		PartitionNum:      2,
 		ReplicationFactor: 4,
 	}
-	manager = newKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg)
-	defer manager.Close()
+	manager = newKafkaTopicManager(topic, changefeedID, adminClient, cfg)
 	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, topic)
 	require.ErrorIs(t, err, errors.ErrKafkaAdminAPI)
 	require.ErrorIs(t, err, sarama.ErrInvalidReplicationFactor)
 	require.NotNil(t, gotFailedTopicDetail)
 	require.Equal(t, "new-topic-failed", gotFailedTopicDetail.Name)
-	require.False(t, gotFailedTopicValidateOnly)
 }
 
 func TestCreateTopicValidatesReplicationFactor(t *testing.T) {
@@ -203,7 +192,6 @@ func TestCreateTopicValidatesReplicationFactor(t *testing.T) {
 	)
 
 	manager := newKafkaTopicManager(
-		context.Background(),
 		topic,
 		common.NewChangefeedID4Test("test", "test"),
 		adminClient,
@@ -214,13 +202,12 @@ func TestCreateTopicValidatesReplicationFactor(t *testing.T) {
 			RequiredAcks:      kafka.WaitForAll,
 		},
 	)
-	defer manager.Close()
 
 	_, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), topic)
 	require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker")
 }
 
-func TestCreateTopicWaitsUntilVisible(t *testing.T) {
+func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) {
 	t.Parallel()
 
 	ctrl := gomock.NewController(t)
@@ -237,14 +224,13 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) {
 			map[string]kafka.TopicDetail{}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
 			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().CreateTopic(gomock.Any(), false).DoAndReturn(
-			func(detail *kafka.TopicDetail, validateOnly bool) error {
+		adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn(
+			func(detail *kafka.TopicDetail) error {
 				require.Equal(t, &kafka.TopicDetail{
 					Name:              topic,
 					NumPartitions:     2,
 					ReplicationFactor: 1,
 				}, detail)
-				require.False(t, validateOnly)
 				return nil
 			}),
 		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
@@ -260,12 +246,35 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) {
 
 	ctx := context.Background()
 	changefeedID := common.NewChangefeedID4Test("test", "test")
-	manager := newKafkaTopicManager(ctx, topic, changefeedID, adminClient, cfg)
-	defer manager.Close()
+	err := EnsureTopic(ctx, changefeedID, topic, cfg, adminClient)
+	require.NoError(t, err)
+}
+
+func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) {
+	t.Parallel()
 
-	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, topic)
+	ctrl := gomock.NewController(t)
+	adminClient := kafka.NewMockClusterAdminClient(ctrl)
+	topic := "existing-topic"
+	adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).Return(
+		map[string]kafka.TopicDetail{
+			topic: {
+				Name:          topic,
+				NumPartitions: 2,
+			},
+		}, nil,
+	)
+
+	manager, err := GetTopicManagerAndTryCreateTopic(
+		t.Context(),
+		common.NewChangefeedID4Test("test", "test"),
+		topic,
+		&kafka.AutoCreateTopicConfig{PartitionNum: 2},
+		adminClient,
+	)
 	require.NoError(t, err)
-	require.Equal(t, int32(2), partitionNum)
+	defer manager.Close()
+	require.NotNil(t, manager.(*kafkaTopicManager).cancel)
 }
 
 func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {
@@ -283,16 +292,16 @@ func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {
 
 	changefeedID := common.NewChangefeedID4Test("test", "test")
 	ctx := context.Background()
-	manager := newKafkaTopicManager(ctx, "precreated-topic", changefeedID, adminClient, cfg)
-	defer manager.Close()
+	defaultTopic := "default-topic"
+	manager := newKafkaTopicManager(defaultTopic, changefeedID, adminClient, cfg)
 
-	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, "precreated-topic")
+	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, defaultTopic)
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 	require.False(t, adminClient.createTopicCalled)
 	require.Equal(t, 2, adminClient.describeCount)
 
-	partitions, ok := manager.topics.Load("precreated-topic")
+	partitions, ok := manager.topics.Load(defaultTopic)
 	require.True(t, ok)
 	require.Equal(t, int32(2), partitions)
 }
@@ -312,16 +321,16 @@ func TestCreateTopicWithCreateDenied(t *testing.T) {
 
 	changefeedID := common.NewChangefeedID4Test("test", "test")
 	ctx := context.Background()
-	manager := newKafkaTopicManager(ctx, "precreated-topic", changefeedID, adminClient, cfg)
-	defer manager.Close()
+	defaultTopic := "default-topic"
+	manager := newKafkaTopicManager(defaultTopic, changefeedID, adminClient, cfg)
 
-	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, "precreated-topic")
+	partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, defaultTopic)
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 	require.True(t, adminClient.createTopicCalled)
 	require.Equal(t, 2, adminClient.describeCount)
 
-	partitions, ok := manager.topics.Load("precreated-topic")
+	partitions, ok := manager.topics.Load(defaultTopic)
 	require.True(t, ok)
 	require.Equal(t, int32(2), partitions)
 }

pkg/sink/kafka/admin.go

@@ -155,13 +155,13 @@ func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]
 	return result, nil
 }
 
-func (a *saramaAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) error {
+func (a *saramaAdminClient) CreateTopic(detail *TopicDetail) error {
 	request := &sarama.TopicDetail{
 		NumPartitions:     detail.NumPartitions,
 		ReplicationFactor: detail.ReplicationFactor,
 	}
 
-	err := a.admin.CreateTopic(detail.Name, request, validateOnly)
+	err := a.admin.CreateTopic(detail.Name, request, false)
 	// Ignore the already exists error because it's not harmful.
 	if err != nil && !strings.Contains(err.Error(), sarama.ErrTopicAlreadyExists.Error()) {
 		return errors.WrapError(errors.ErrKafkaAdminAPI, err, "create-topic", detail.Name)

pkg/sink/kafka/admin_test.go

@@ -61,6 +61,28 @@ func TestGetBrokerConfig(t *testing.T) {
 	})
 }
 
+func TestCreateTopic(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	admin := NewMocksaramaClusterAdmin(ctrl)
+	admin.EXPECT().CreateTopic("test-topic", &sarama.TopicDetail{
+		NumPartitions:     3,
+		ReplicationFactor: 2,
+	}, false).Return(nil)
+
+	client := &saramaAdminClient{
+		changefeed: common.NewChangeFeedIDWithName("test", "default"),
+		admin:      admin,
+	}
+	err := client.CreateTopic(&TopicDetail{
+		Name:              "test-topic",
+		NumPartitions:     3,
+		ReplicationFactor: 2,
+	})
+	require.NoError(t, err)
+}
+
 func TestAdminClientClose(t *testing.T) {
 	tests := []struct {
 		name  string