Skip to content

kafka: avoid create changefeed failures if can't get a topic from broker (#5696) #5810

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ti-chi-bot wants to merge 2 commits into from
Closed

kafka: avoid create changefeed failures if can't get a topic from broker (#5696) #5810

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4fb0277
This is an automated cherry-pick of #5696
wk989898 Jul 21, 2026
ff08613
Merge branch 'release-8.5' into cherry-pick-5696-to-release-8.5
wk989898 Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions downstreamadapter/sink/topicmanager/kafka_topic_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
if kafka.IsAdminAuthorizationFailed(err) {
return m.useConfiguredPartitionNum(topicName, err), nil
}
return 0, err
return 0, errors.Trace(err)
}
if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok {
return numPartition, nil
Expand All @@ -291,7 +291,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
if kafka.IsAdminAuthorizationFailed(err) {
return m.useConfiguredPartitionNum(topicName, err), nil
}
return 0, err
return 0, errors.Trace(err)
}

err = m.waitUntilTopicVisible(ctx, topicName)
Expand Down Expand Up @@ -328,11 +328,11 @@ func (m *kafkaTopicManager) tryStoreTopicMeta(
}

func (m *kafkaTopicManager) useConfiguredPartitionNum(topicName string, cause error) int32 {
log.Warn("kafka topic creation skipped due to authorization failure",
log.Warn("skip Kafka topic creation because topic authorization failed",
zap.String("keyspace", m.changefeedID.Keyspace()),
zap.String("changefeed", m.changefeedID.Name()),
zap.String("topic", topicName),
zap.Int32("partitionNum", m.cfg.PartitionNum),
zap.Int32("partitionNumber", m.cfg.PartitionNum),
zap.Error(cause))
m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum)
return m.cfg.PartitionNum
Expand Down
67 changes: 64 additions & 3 deletions downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@

changefeedID := common.NewChangefeedID4Test("test", "test")
ctx := context.Background()
manager := newKafkaTopicManager(ctx, kafka.DefaultMockTopicName, changefeedID, adminClient, cfg)

Check failure on line 91 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

undefined: kafka.DefaultMockTopicName
var gotNewTopicDetail *kafka.TopicDetail
var gotNewTopicValidateOnly bool
var gotFailedTopicDetail *kafka.TopicDetail
Expand Down Expand Up @@ -129,11 +130,11 @@
func(detail *kafka.TopicDetail, validateOnly bool) error {
gotFailedTopicDetail = detail
gotFailedTopicValidateOnly = validateOnly
return errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidReplicationFactor, "create-topic", detail.Name)
return sarama.ErrInvalidReplicationFactor
}),
)

manager := newKafkaTopicManager(ctx, kafkaTopicManagerTestTopic, changefeedID, adminClient, cfg)

Check failure on line 137 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

no new variables on left side of :=
defer manager.Close()
partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, kafkaTopicManagerTestTopic)
require.NoError(t, err)
Expand Down Expand Up @@ -231,7 +232,7 @@
ReplicationFactor: 1,
}

topic := "delayed-topic"
topic := "new_topic"
gomock.InOrder(
adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).Return(
map[string]kafka.TopicDetail{}, nil),
Expand All @@ -248,7 +249,9 @@
return nil
}),
adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
map[string]kafka.TopicDetail{}, nil),
nil, sarama.ErrUnknownTopicOrPartition),
adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
nil, sarama.ErrUnknownTopicOrPartition),
adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
map[string]kafka.TopicDetail{
topic: {
Expand All @@ -268,7 +271,7 @@
require.Equal(t, int32(2), partitionNum)
}

func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {

Check failure on line 274 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

other declaration of TestCreateTopicWithTopicDescribeDenied
t.Parallel()

ctrl := gomock.NewController(t)
Expand Down Expand Up @@ -297,7 +300,7 @@
require.Equal(t, int32(2), partitions)
}

func TestCreateTopicWithCreateDenied(t *testing.T) {

Check failure on line 303 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

other declaration of TestCreateTopicWithCreateDenied
t.Parallel()

ctrl := gomock.NewController(t)
Expand Down Expand Up @@ -325,3 +328,61 @@
require.True(t, ok)
require.Equal(t, int32(2), partitions)
}

func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {

Check failure on line 332 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

TestCreateTopicWithTopicDescribeDenied redeclared in this block
t.Parallel()

ctrl := gomock.NewController(t)
adminClient := &mockAdminClientWithDeniedDescribe{
MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl),
}
cfg := &kafka.AutoCreateTopicConfig{
AutoCreate: true,
PartitionNum: 2,
ReplicationFactor: 1,
}

changefeedID := common.NewChangefeedID4Test("test", "test")
ctx := context.Background()
manager := newKafkaTopicManager(ctx, "precreated-topic", changefeedID, adminClient, cfg)
defer manager.Close()

partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, "precreated-topic")
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")
require.True(t, ok)
require.Equal(t, int32(2), partitions)
}

func TestCreateTopicWithCreateDenied(t *testing.T) {

Check failure on line 361 in downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

View workflow job for this annotation

GitHub Actions / Classic Unit Tests 

TestCreateTopicWithCreateDenied redeclared in this block
t.Parallel()

ctrl := gomock.NewController(t)
adminClient := &mockAdminClientWithDeniedCreate{
MockClusterAdminClient: kafka.NewMockClusterAdminClient(ctrl),
}
cfg := &kafka.AutoCreateTopicConfig{
AutoCreate: true,
PartitionNum: 2,
ReplicationFactor: 1,
}

changefeedID := common.NewChangefeedID4Test("test", "test")
ctx := context.Background()
manager := newKafkaTopicManager(ctx, "precreated-topic", changefeedID, adminClient, cfg)
defer manager.Close()

partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, "precreated-topic")
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")
require.True(t, ok)
require.Equal(t, int32(2), partitions)
}
Loading