diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index c6963573d8..291aef0596 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/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() diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 49231e3116..9094b5758b 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/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() diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 41298460d8..8551b25020 100644 --- a/pkg/sink/kafka/admin.go +++ b/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 { diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 6a5624427f..84abedae7e 100644 --- a/pkg/sink/kafka/options.go +++ b/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 {