From 21116f6d8c0ab9cc6433f9de6c1cdbf01d00de34 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 6 Aug 2026 16:17:36 +0800 Subject: [PATCH 1/4] verify also create topic and wait it --- downstreamadapter/sink/kafka/sink.go | 22 +---- .../sink/topicmanager/kafka_topic_manager.go | 57 +++++++------ .../topicmanager/kafka_topic_manager_test.go | 81 ++++++++++--------- pkg/sink/kafka/admin.go | 4 +- pkg/sink/kafka/admin_test.go | 22 +++++ pkg/sink/kafka/cluster_admin_client.go | 2 +- pkg/sink/kafka/cluster_admin_client_mock.go | 8 +- pkg/sink/kafka/options_test.go | 6 +- 8 files changed, 112 insertions(+), 90 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 195a7e3c25..f6351fb269 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..ffa1bd7077 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, @@ -57,38 +85,19 @@ func GetTopicManagerAndTryCreateTopic( adminClient kafka.ClusterAdminClient, ) (TopicManager, error) { topicManager := newKafkaTopicManager( - ctx, topic, changefeedID, adminClient, topicCfg, + topic, changefeedID, adminClient, topicCfg, ) if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil { return nil, err } + refreshCtx, cancel := context.WithCancel(ctx) + topicManager.cancel = cancel + go topicManager.backgroundRefreshMeta(refreshCtx) 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 +247,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", 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 7f49c29402..9c0ca3c6ed 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 } @@ -485,7 +485,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) From d0525b5befe94166a1767453adb14264623e5501 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 6 Aug 2026 16:51:27 +0800 Subject: [PATCH 2/4] adjust the code --- .../sink/topicmanager/kafka_topic_manager.go | 14 ++++-- .../topicmanager/kafka_topic_manager_test.go | 49 +++++++++++++++++++ 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index ffa1bd7077..007059de91 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -72,7 +72,7 @@ func EnsureTopic( adminClient kafka.ClusterAdminClient, ) error { topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg) - _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic) + _, err := topicManager.createTopicAndWaitUntilVisible(ctx, topic, false) return err } @@ -270,13 +270,19 @@ func (m *kafkaTopicManager) createTopic( // CreateTopicAndWaitUntilVisible wraps createTopic and waitUntilTopicVisible together. func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( ctx context.Context, topicName string, +) (int32, error) { + return m.createTopicAndWaitUntilVisible(ctx, topicName, true) +} + +func (m *kafkaTopicManager) createTopicAndWaitUntilVisible( + ctx context.Context, topicName string, allowAuthorizationFailure bool, ) (int32, error) { // If the topic is not in the cache, we try to get the metadata of the topic. // ignoreTopicErr is set to true to ignore the error if the topic is not found, // which means we should create the topic later. topicDetails, err := m.admin.GetTopicsMeta([]string{topicName}, true) if err != nil { - if kafka.IsAdminAuthorizationFailed(err) { + if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } return 0, err @@ -287,7 +293,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( topicDetails, err = m.admin.GetTopicsMeta([]string{topicName}, false) if err != nil { - if kafka.IsAdminAuthorizationFailed(err) { + if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } } else if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok { @@ -297,7 +303,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( start := time.Now() partitionNum, err := m.createTopic(ctx, topicName) if err != nil { - if kafka.IsAdminAuthorizationFailed(err) { + if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } return 0, err diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index b00bf947d0..14dc18498a 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -250,6 +250,55 @@ func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) { require.NoError(t, err) } +func TestEnsureTopicReturnsAuthorizationErrors(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + adminClient func(*gomock.Controller) kafka.ClusterAdminClient + expectedErr error + }{ + { + name: "describe denied", + adminClient: func(ctrl *gomock.Controller) kafka.ClusterAdminClient { + return &mockAdminClientWithDeniedDescribe{ + MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), + } + }, + expectedErr: sarama.ErrTopicAuthorizationFailed, + }, + { + name: "create denied", + adminClient: func(ctrl *gomock.Controller) kafka.ClusterAdminClient { + return &mockAdminClientWithDeniedCreate{ + MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), + } + }, + expectedErr: sarama.ErrClusterAuthorizationFailed, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + cfg := &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + } + err := EnsureTopic( + t.Context(), + common.NewChangefeedID4Test("test", "test"), + "default-topic", + cfg, + tt.adminClient(gomock.NewController(t)), + ) + require.ErrorIs(t, err, tt.expectedErr) + }) + } +} + func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) { t.Parallel() From ff7df374bab0ef2827d6758bd96283f759af2a3b Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 6 Aug 2026 17:11:20 +0800 Subject: [PATCH 3/4] adjust code --- .../sink/topicmanager/kafka_topic_manager.go | 17 +++---- .../topicmanager/kafka_topic_manager_test.go | 49 ------------------- 2 files changed, 7 insertions(+), 59 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 007059de91..5618cb21ee 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -72,7 +72,7 @@ func EnsureTopic( adminClient kafka.ClusterAdminClient, ) error { topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg) - _, err := topicManager.createTopicAndWaitUntilVisible(ctx, topic, false) + _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic) return err } @@ -268,21 +268,18 @@ 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) { - return m.createTopicAndWaitUntilVisible(ctx, topicName, true) -} - -func (m *kafkaTopicManager) createTopicAndWaitUntilVisible( - ctx context.Context, topicName string, allowAuthorizationFailure bool, ) (int32, error) { // If the topic is not in the cache, we try to get the metadata of the topic. // ignoreTopicErr is set to true to ignore the error if the topic is not found, // which means we should create the topic later. topicDetails, err := m.admin.GetTopicsMeta([]string{topicName}, true) if err != nil { - if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { + if kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } return 0, err @@ -293,7 +290,7 @@ func (m *kafkaTopicManager) createTopicAndWaitUntilVisible( topicDetails, err = m.admin.GetTopicsMeta([]string{topicName}, false) if err != nil { - if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { + if kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } } else if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok { @@ -303,7 +300,7 @@ func (m *kafkaTopicManager) createTopicAndWaitUntilVisible( start := time.Now() partitionNum, err := m.createTopic(ctx, topicName) if err != nil { - if allowAuthorizationFailure && kafka.IsAdminAuthorizationFailed(err) { + if kafka.IsAdminAuthorizationFailed(err) { return m.useConfiguredPartitionNum(topicName, err), nil } return 0, err diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 14dc18498a..b00bf947d0 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -250,55 +250,6 @@ func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) { require.NoError(t, err) } -func TestEnsureTopicReturnsAuthorizationErrors(t *testing.T) { - t.Parallel() - - tests := []struct { - name string - adminClient func(*gomock.Controller) kafka.ClusterAdminClient - expectedErr error - }{ - { - name: "describe denied", - adminClient: func(ctrl *gomock.Controller) kafka.ClusterAdminClient { - return &mockAdminClientWithDeniedDescribe{ - MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), - } - }, - expectedErr: sarama.ErrTopicAuthorizationFailed, - }, - { - name: "create denied", - adminClient: func(ctrl *gomock.Controller) kafka.ClusterAdminClient { - return &mockAdminClientWithDeniedCreate{ - MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl), - } - }, - expectedErr: sarama.ErrClusterAuthorizationFailed, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - cfg := &kafka.AutoCreateTopicConfig{ - AutoCreate: true, - PartitionNum: 2, - ReplicationFactor: 1, - } - err := EnsureTopic( - t.Context(), - common.NewChangefeedID4Test("test", "test"), - "default-topic", - cfg, - tt.adminClient(gomock.NewController(t)), - ) - require.ErrorIs(t, err, tt.expectedErr) - }) - } -} - func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) { t.Parallel() From 7d948647cd04f6a908e74499eeff879b1624fece Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 7 Aug 2026 09:53:32 +0800 Subject: [PATCH 4/4] simplify the code further --- .../sink/topicmanager/kafka_topic_manager.go | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 5618cb21ee..e436e8801a 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -84,16 +84,14 @@ func GetTopicManagerAndTryCreateTopic( topicCfg *kafka.AutoCreateTopicConfig, adminClient kafka.ClusterAdminClient, ) (TopicManager, error) { - topicManager := newKafkaTopicManager( - topic, changefeedID, adminClient, topicCfg, - ) + topicManager := newKafkaTopicManager(topic, changefeedID, adminClient, topicCfg) if _, err := topicManager.CreateTopicAndWaitUntilVisible(ctx, topic); err != nil { return nil, err } - refreshCtx, cancel := context.WithCancel(ctx) + ctx, cancel := context.WithCancel(ctx) topicManager.cancel = cancel - go topicManager.backgroundRefreshMeta(refreshCtx) + go topicManager.backgroundRefreshMeta(ctx) return topicManager, nil }