kafka: use default option when meeting authorization and ACLs fails (#5562)

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -296,19 +296,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)
 	}
 
@@ -320,6 +330,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

@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/IBM/sarama"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/sink/kafka"
 	"github.com/stretchr/testify/require"
@@ -43,6 +44,53 @@ func (m *mockAdminClientForHeartbeat) GetHeartbeatCount() int {
 	return m.heartbeatCount
 }
 
+type mockAdminClientWithDeniedDescribe struct {
+	*kafka.ClusterAdminClientMockImpl
+	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.ClusterAdminClientMockImpl
+	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 TestKafkaTopicManagerHeartbeat(t *testing.T) {
 	t.Parallel()
 
@@ -127,6 +175,64 @@ func TestCreateTopic(t *testing.T) {
 	)
 }
 
+func TestCreateTopicWithTopicDescribeDenied(t *testing.T) {
+	t.Parallel()
+
+	adminClient := &mockAdminClientWithDeniedDescribe{
+		ClusterAdminClientMockImpl: kafka.NewClusterAdminClientMockImpl(),
+	}
+	defer adminClient.Close()
+	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()
+
+	adminClient := &mockAdminClientWithDeniedCreate{
+		ClusterAdminClientMockImpl: kafka.NewClusterAdminClientMockImpl(),
+	}
+	defer adminClient.Close()
+	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)
+}
+
 func TestCreateTopicWithDelay(t *testing.T) {
 	t.Parallel()
 

pkg/sink/kafka/admin.go

@@ -153,6 +153,16 @@ 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 isKafkaError(err, sarama.ErrTopicAuthorizationFailed) ||
+		isKafkaError(err, sarama.ErrClusterAuthorizationFailed)
+}
+
+func isKafkaError(err error, target sarama.KError) bool {
+	return errors.Is(err, target) || errors.Cause(err) == target
+}
+
 func (a *saramaAdminClient) GetTopicsPartitionsNum(topics []string) (map[string]int32, error) {
 	result := make(map[string]int32, len(topics))
 	for _, topic := range topics {

pkg/sink/kafka/options.go

@@ -618,12 +618,15 @@ func adjustOptions(
 			TopicMaxMessageBytesConfigName,
 			BrokerMessageMaxBytesConfigName,
 		)
+		var topicMaxMessageBytes int
 		if err != nil {
-			return errors.Trace(err)
-		}
-		topicMaxMessageBytes, err := strconv.Atoi(topicMaxMessageBytesStr)
-		if err != nil {
-			return errors.Trace(err)
+			log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default")
+			topicMaxMessageBytes = options.MaxMessageBytes
+		} else {
+			topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr)
+			if err != nil {
+				return errors.Trace(err)
+			}
 		}
 
 		maxMessageBytes := topicMaxMessageBytes - maxMessageBytesOverhead
@@ -654,14 +657,16 @@ func adjustOptions(
 		return nil
 	}
 
+	var brokerMessageMaxBytes int
 	brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName)
 	if err != nil {
-		log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration")
-		return errors.Trace(err)
-	}
-	brokerMessageMaxBytes, err := strconv.Atoi(brokerMessageMaxBytesStr)
-	if err != nil {
-		return errors.Trace(err)
+		log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default")
+		brokerMessageMaxBytes = options.MaxMessageBytes
+	} else {
+		brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr)
+		if err != nil {
+			return errors.Trace(err)
+		}
 	}
 
 	// when create the topic, `max.message.bytes` is decided by the broker,
@@ -728,9 +733,9 @@ func validateMinInsyncReplicas(
 				"to the minimum number of in-sync replicas" +
 				"if you want to use `required-acks` = -1." +
 				"Otherwise, TiCDC will not be able to send messages to the topic.")
-			return nil
 		}
-		return err
+		log.Warn("TiCDC meets error when get `min.insync.replicas` from broker's configuration, assume the config is valid")
+		return nil
 	}
 	minInsyncReplicas, err := strconv.Atoi(minInsyncReplicasStr)
 	if err != nil {