diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 2133e1c6ba..6c6332a4ae 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/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 { diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 3db0938012..e436e8801a 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/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) { diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 3708d77668..b00bf947d0 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/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) } diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 98e3b154d1..8ad44cf5ea 100644 --- a/pkg/sink/kafka/admin.go +++ b/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) diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 3bcd3d0468..a79ad5b06a 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/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 diff --git a/pkg/sink/kafka/cluster_admin_client.go b/pkg/sink/kafka/cluster_admin_client.go index 4f1ff36996..d378658c3a 100644 --- a/pkg/sink/kafka/cluster_admin_client.go +++ b/pkg/sink/kafka/cluster_admin_client.go @@ -45,7 +45,7 @@ type ClusterAdminClient interface { GetTopicsPartitionsNum(topics []string) (map[string]int32, error) // CreateTopic creates a new topic. - CreateTopic(detail *TopicDetail, validateOnly bool) error + CreateTopic(detail *TopicDetail) error // Close shuts down the admin client. Close() diff --git a/pkg/sink/kafka/cluster_admin_client_mock.go b/pkg/sink/kafka/cluster_admin_client_mock.go index dfeebbd773..4fb06a2a56 100644 --- a/pkg/sink/kafka/cluster_admin_client_mock.go +++ b/pkg/sink/kafka/cluster_admin_client_mock.go @@ -46,17 +46,17 @@ func (mr *MockClusterAdminClientMockRecorder) Close() *gomock.Call { } // CreateTopic mocks base method. -func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail, validateOnly bool) error { +func (m *MockClusterAdminClient) CreateTopic(detail *TopicDetail) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "CreateTopic", detail, validateOnly) + ret := m.ctrl.Call(m, "CreateTopic", detail) ret0, _ := ret[0].(error) return ret0 } // CreateTopic indicates an expected call of CreateTopic. -func (mr *MockClusterAdminClientMockRecorder) CreateTopic(detail, validateOnly interface{}) *gomock.Call { +func (mr *MockClusterAdminClientMockRecorder) CreateTopic(detail interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockClusterAdminClient)(nil).CreateTopic), detail, validateOnly) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTopic", reflect.TypeOf((*MockClusterAdminClient)(nil).CreateTopic), detail) } // GetAllBrokers mocks base method. diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 025df9086b..93498b384d 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -76,7 +76,7 @@ func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { DoAndReturn(fixture.getBrokerConfig).AnyTimes() fixture.admin.EXPECT().GetTopicConfig(gomock.Any(), gomock.Any()). DoAndReturn(fixture.getTopicConfig).AnyTimes() - fixture.admin.EXPECT().CreateTopic(gomock.Any(), gomock.Any()). + fixture.admin.EXPECT().CreateTopic(gomock.Any()). DoAndReturn(fixture.createTopic).AnyTimes() return fixture @@ -127,7 +127,7 @@ func (f *kafkaAdminFixture) getTopicConfig(topicName string, configName string) return "", false, nil } -func (f *kafkaAdminFixture) createTopic(detail *TopicDetail, _ bool) error { +func (f *kafkaAdminFixture) createTopic(detail *TopicDetail) error { if detail.ReplicationFactor > mockClusterReplicationFactor { return sarama.ErrInvalidReplicationFactor } @@ -486,7 +486,7 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t Name: topicName, NumPartitions: 3, } - err := adminClient.CreateTopic(detail, false) + err := adminClient.CreateTopic(detail) require.NoError(t, err) configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture)