kafka: avoid create changefeed failures if can't get a topic from broker

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -287,19 +287,29 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
 	// which means we should create the topic later.
 	topicDetails, err := m.admin.GetTopicsMeta([]string{topicName}, true)
 	if err != nil {
+		if kafka.IsAdminAuthorizationFailed(err) {
+			return m.useConfiguredPartitionNum(topicName, err), nil
+		}
 		return 0, errors.Trace(err)
 	}
-	if detail, ok := topicDetails[topicName]; ok {
-		numPartition := detail.NumPartitions
-		if topicName == m.defaultTopic {
-			numPartition = m.cfg.PartitionNum
+	if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok {
+		return numPartition, nil
+	}
+
+	topicDetails, err = m.admin.GetTopicsMeta([]string{topicName}, false)
+	if err != nil {
+		if kafka.IsAdminAuthorizationFailed(err) {
+			return m.useConfiguredPartitionNum(topicName, err), nil
 		}
-		m.tryUpdatePartitionsAndLogging(topicName, numPartition)
+	} else if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok {
 		return numPartition, nil
 	}
 
 	partitionNum, err := m.createTopic(ctx, topicName)
 	if err != nil {
+		if kafka.IsAdminAuthorizationFailed(err) {
+			return m.useConfiguredPartitionNum(topicName, err), nil
+		}
 		return 0, errors.Trace(err)
 	}
 
@@ -311,6 +321,32 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible(
 	return partitionNum, nil
 }
 
+func (m *kafkaTopicManager) tryStoreTopicMeta(
+	topicName string, topicDetails map[string]kafka.TopicDetail,
+) (int32, bool) {
+	detail, ok := topicDetails[topicName]
+	if !ok {
+		return 0, false
+	}
+	numPartition := detail.NumPartitions
+	if topicName == m.defaultTopic {
+		numPartition = m.cfg.PartitionNum
+	}
+	m.tryUpdatePartitionsAndLogging(topicName, numPartition)
+	return numPartition, true
+}
+
+func (m *kafkaTopicManager) useConfiguredPartitionNum(topicName string, cause error) int32 {
+	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("partitionNumber", m.cfg.PartitionNum),
+		zap.Error(cause))
+	m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum)
+	return m.cfg.PartitionNum
+}
+
 // Close exits the background goroutine.
 func (m *kafkaTopicManager) Close() {
 	m.cancel()

downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

@@ -26,6 +26,53 @@ import (
 
 const kafkaTopicManagerTestTopic = "mock_topic"
 
+type mockAdminClientWithDeniedDescribe struct {
+	*kafka.MockClusterAdminClient
+	createTopicCalled bool
+	describeCount     int
+}
+
+func (m *mockAdminClientWithDeniedDescribe) GetTopicsMeta(
+	topics []string,
+	ignoreTopicError bool,
+) (map[string]kafka.TopicDetail, error) {
+	m.describeCount++
+	if ignoreTopicError {
+		return map[string]kafka.TopicDetail{}, nil
+	}
+	return nil, sarama.ErrTopicAuthorizationFailed
+}
+
+func (m *mockAdminClientWithDeniedDescribe) CreateTopic(
+	detail *kafka.TopicDetail,
+	validateOnly bool,
+) error {
+	m.createTopicCalled = true
+	return nil
+}
+
+type mockAdminClientWithDeniedCreate struct {
+	*kafka.MockClusterAdminClient
+	createTopicCalled bool
+	describeCount     int
+}
+
+func (m *mockAdminClientWithDeniedCreate) GetTopicsMeta(
+	topics []string,
+	ignoreTopicError bool,
+) (map[string]kafka.TopicDetail, error) {
+	m.describeCount++
+	return map[string]kafka.TopicDetail{}, nil
+}
+
+func (m *mockAdminClientWithDeniedCreate) CreateTopic(
+	detail *kafka.TopicDetail,
+	validateOnly bool,
+) error {
+	m.createTopicCalled = true
+	return sarama.ErrClusterAuthorizationFailed
+}
+
 func TestCreateTopic(t *testing.T) {
 	t.Parallel()
 
@@ -53,6 +100,8 @@ func TestCreateTopic(t *testing.T) {
 			}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, true).Return(
 			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 {
 				gotNewTopicDetail = detail
@@ -68,8 +117,12 @@ func TestCreateTopic(t *testing.T) {
 			}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic2"}, true).Return(
 			map[string]kafka.TopicDetail{}, nil),
+		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic2"}, false).Return(
+			map[string]kafka.TopicDetail{}, nil),
 		adminClient.EXPECT().GetTopicsMeta([]string{"new-topic-failed"}, true).Return(
 			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 {
 				gotFailedTopicDetail = detail
@@ -144,6 +197,8 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) {
 	gomock.InOrder(
 		adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).Return(
 			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 {
 				require.Equal(t, &kafka.TopicDetail{
@@ -176,3 +231,61 @@ func TestCreateTopicWaitsUntilVisible(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 }
+
+func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {
+	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) {
+	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)
+}

pkg/sink/kafka/admin.go

@@ -153,6 +153,12 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool
 	return result, nil
 }
 
+// IsAdminAuthorizationFailed checks whether err is an authorization failure from Kafka admin APIs.
+func IsAdminAuthorizationFailed(err error) bool {
+	return errors.Is(err, sarama.ErrTopicAuthorizationFailed) ||
+		errors.Is(err, sarama.ErrClusterAuthorizationFailed)
+}
+
 func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) {
 	result := make(map[string]int32, len(topics))
 	for _, topic := range topics {