From 870f3ca9c3485c83ad739f7f2f20c68f3d868134 Mon Sep 17 00:00:00 2001 From: Ling Jin <7138436+3AceShowHand@users.noreply.github.com> Date: Sat, 29 Aug 2026 00:12:47 +0800 Subject: [PATCH] This is an automated cherry-pick of #6081 Signed-off-by: ti-chi-bot --- downstreamadapter/sink/kafka/sink.go | 14 + downstreamadapter/sink/kafka/sink_test.go | 423 +++++++++++++++++ .../sink/topicmanager/kafka_topic_manager.go | 19 + .../topicmanager/kafka_topic_manager_test.go | 235 +++++++++ pkg/sink/kafka/admin.go | 23 +- pkg/sink/kafka/options.go | 2 + pkg/sink/kafka/sarama_admin_test.go | 449 ++++++++++++++++++ 7 files changed, 1162 insertions(+), 3 deletions(-) create mode 100644 pkg/sink/kafka/sarama_admin_test.go diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index ab3590c6a7..b24cb370eb 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -141,6 +141,9 @@ func (s *sink) IsNormal() bool { } func (s *sink) AddDMLEvent(event *commonEvent.DMLEvent) { + if !s.isNormal.Load() { + return + } s.eventChan.Push(event) } @@ -169,6 +172,7 @@ func (s *sink) WriteBlockEvent(event commonEvent.BlockEvent) error { } func (s *sink) close() { + s.isNormal.Store(false) s.eventChan.Close() s.rowChan.Close() } @@ -227,6 +231,11 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { if err != nil { return errors.Trace(err) } + select { + case <-ctx.Done(): + return context.Cause(ctx) + default: + } s.rowChan.Push(events...) } } @@ -512,7 +521,12 @@ func (s *sink) getAllTableNames(ts uint64) []*commonEvent.SchemaTableName { return s.tableSchemaStore.GetAllTableNames(ts, true) } +<<<<<<< HEAD func (s *sink) Close(_ bool) { +======= +func (s *sink) Close() { + s.close() +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) s.ddlProducer.Close() s.dmlProducer.Close() s.comp.close() diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 2e02784f5e..30f4c7724c 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -132,12 +132,24 @@ func TestDDLProducerHeartbeat(t *testing.T) { _, err := newKafkaSinkForTestWithProducers(ctx, nil, producer) require.NoError(t, err) +<<<<<<< HEAD // Wait for a sufficient amount of time to ensure the heartbeat ticker triggers several times. // Waiting for 11 seconds to allow for at least two heartbeats. // Use Eventually to avoid test flakiness. require.Eventually(t, func() bool { return producer.GetHeartbeatCount() >= 2 }, 11*time.Second, 150*time.Millisecond, "Heartbeat should be called periodically") +======= + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + factory := kafka.NewMockFactory(ctrl) + gomock.InOrder( + factory.EXPECT().AdminClient(gomock.Any()).Return(adminClient, nil), + adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, false).Return( + map[string]kafka.TopicDetail{kafkaSinkTestTopic: {Name: kafkaSinkTestTopic}}, nil), + adminClient.EXPECT().Close(), + ) +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) // Verify that closing the manager stops the heartbeat. countBeforeClose := producer.GetHeartbeatCount() @@ -266,3 +278,414 @@ func TestKafkaSinkBasicFunctionality(t *testing.T) { cancel() kafkaSink.AddCheckpointTs(12345) } +<<<<<<< HEAD +======= + +func TestKafkaSinkBatchConfig(t *testing.T) { + sink := &sink{} + require.Equal(t, 4096, sink.BatchCount()) + require.Zero(t, sink.BatchBytes()) +} + +func TestKafkaSinkConstructionAndCleanup(t *testing.T) { + t.Run("async producer creation fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(nil, cause) + gomock.InOrder( + adminClient.EXPECT().Close(), + topicManager.EXPECT().Close(), + ) + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "async-creation-fails"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.Nil(t, kafkaSink) + require.Equal(t, cause, err) + }) + + t.Run("sync producer creation fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(nil, cause) + gomock.InOrder( + asyncProducer.EXPECT().Close(), + adminClient.EXPECT().Close(), + topicManager.EXPECT().Close(), + ) + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "sync-creation-fails"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.Nil(t, kafkaSink) + require.Equal(t, cause, err) + }) + + t.Run("successful construction owns resources until close", func(t *testing.T) { + ctrl := gomock.NewController(t) + factory := kafka.NewMockFactory(ctrl) + adminClient := kafka.NewMockAdminClient(ctrl) + topicManager := topicmanager.NewMockTopicManager(ctrl) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + var closeCount atomic.Int64 + + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) + factory.EXPECT().MetricsCollector(adminClient).Return(noopMetricsCollector{}) + gomock.InOrder( + syncProducer.EXPECT().Close().Do(func() { closeCount.Add(1) }), + asyncProducer.EXPECT().Close().Do(func() { closeCount.Add(1) }), + adminClient.EXPECT().Close().Do(func() { closeCount.Add(1) }), + topicManager.EXPECT().Close().Do(func() { closeCount.Add(1) }), + ) + + kafkaSink, err := newWithComponents( + t.Context(), + common.NewChangefeedID4Test("test", "successful-construction"), + common.DefaultKeyspaceID, + config.ProtocolOpen, + components{factory: factory, adminClient: adminClient, topicManager: topicManager}, + ) + + require.NoError(t, err) + require.Zero(t, closeCount.Load()) + require.True(t, kafkaSink.IsNormal()) + + kafkaSink.Close() + require.Equal(t, int64(4), closeCount.Load()) + require.False(t, kafkaSink.IsNormal()) + kafkaSink.AddDMLEvent(&commonEvent.DMLEvent{}) + require.Zero(t, kafkaSink.eventChan.Len()) + + _, ok, err := kafkaSink.eventChan.GetWithContext(t.Context()) + require.NoError(t, err) + require.False(t, ok) + _, ok, err = kafkaSink.rowChan.GetWithContext(t.Context()) + require.NoError(t, err) + require.False(t, ok) + }) +} + +func TestKafkaSinkDML(t *testing.T) { + eventHelper := commonEvent.NewEventTestHelper(t) + defer eventHelper.Close() + eventHelper.Tk().MustExec("use test") + require.NotNil(t, eventHelper.DDL2Job("create table t (id int primary key, name varchar(32))")) + + t.Run("routes DML event and forwards producer callback", func(t *testing.T) { + var callbackCount atomic.Int64 + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (1, 'one')") + dmlEvent.PostTxnFlushed = []func(){func() { callbackCount.Add(1) }} + + sent := make(chan *codecCommon.Message, 1) + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, topicManager, asyncProducer, _ := newKafkaSinkForTest( + t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil) + asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()). + DoAndReturn(func(_ context.Context, _ string, _ int32, message *codecCommon.Message) error { + sent <- message + return nil + }) + + runDone := make(chan error, 1) + go func() { runDone <- kafkaSink.sendDMLEvent(ctx) }() + kafkaSink.AddDMLEvent(dmlEvent) + + select { + case message := <-sent: + require.NotEmpty(t, message.Key) + require.NotEmpty(t, message.Value) + require.Equal(t, 1, message.GetRowsCount()) + require.NotNil(t, message.Callback) + require.Zero(t, callbackCount.Load()) + message.Callback() + require.Equal(t, int64(1), callbackCount.Load()) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for Kafka Sink to send the DML message") + } + + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + select { + case err := <-runDone: + require.Equal(t, cause, err) + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for Kafka Sink workers to exit") + } + require.Equal(t, int64(1), callbackCount.Load()) + }) + + t.Run("returns AsyncSend error unchanged", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (2, 'two')") + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + kafkaSink, topicManager, asyncProducer, _ := newKafkaSinkForTest( + t, ctx, config.ProtocolCanalJSON, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(1), nil) + asyncProducer.EXPECT().AsyncSend(gomock.Any(), kafkaSinkTestTopic, int32(0), gomock.Any()).Return(cause) + + kafkaSink.AddDMLEvent(dmlEvent) + err := kafkaSink.sendDMLEvent(ctx) + + require.Equal(t, cause, err) + }) + + t.Run("returns topic manager error unchanged", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (3, 'three')") + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + kafkaSink.AddDMLEvent(dmlEvent) + + require.Equal(t, cause, kafkaSink.calculateKeyPartitions(t.Context())) + }) + + t.Run("canceled after topic lookup", func(t *testing.T) { + dmlEvent := eventHelper.DML2Event("test", "t", "insert into t values (4, 'four')") + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic). + DoAndReturn(func(context.Context, string) (int32, error) { + cancel(cause) + return 1, nil + }) + kafkaSink.AddDMLEvent(dmlEvent) + require.Equal(t, cause, kafkaSink.calculateKeyPartitions(ctx)) + require.Zero(t, kafkaSink.rowChan.Len()) + }) +} + +func TestKafkaSinkDDL(t *testing.T) { + ddlEvent := &commonEvent.DDLEvent{ + Type: byte(model.ActionCreateTable), + SchemaName: "test", + TableName: "t", + Query: "create table test.t (id int primary key)", + FinishedTs: 1, + } + + t.Run("all partitions", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(4), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Key) + require.NotEmpty(t, message.Value) + return nil + }) + + require.NoError(t, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("partition zero", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolCanalJSON, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(4), nil) + syncProducer.EXPECT().SendMessage(kafkaSinkTestTopic, int32(0), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Value) + return nil + }) + + require.NoError(t, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("topic manager error", func(t *testing.T) { + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + + require.Equal(t, cause, kafkaSink.sendDDLEvent(ddlEvent)) + }) + + t.Run("producer error marks sink abnormal", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(2), nil) + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(2), gomock.Any()).Return(cause) + + require.Equal(t, cause, kafkaSink.WriteBlockEvent(ddlEvent)) + require.False(t, kafkaSink.IsNormal()) + }) + + t.Run("nil encoded message", func(t *testing.T) { + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolDebezium, &config.SinkConfig{}) + unsupportedDDL := &commonEvent.DDLEvent{Type: byte(model.ActionNone), Query: "unsupported"} + + require.NoError(t, kafkaSink.sendDDLEvent(unsupportedDDL)) + }) + + t.Run("unsupported block event", func(t *testing.T) { + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + syncPoint := commonEvent.NewSyncPointEvent(common.NewDispatcherID(), 1, 1, 1) + + require.ErrorIs(t, kafkaSink.WriteBlockEvent(syncPoint), errors.ErrInvalidEventType) + }) +} + +func TestKafkaSinkCheckpoint(t *testing.T) { + t.Run("default topic without tables", func(t *testing.T) { + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(3), nil) + syncProducer.EXPECT().SendMessages(kafkaSinkTestTopic, int32(3), gomock.Any()). + DoAndReturn(func(_ string, _ int32, message *codecCommon.Message) error { + require.NotEmpty(t, message.Key) + return nil + }) + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("all active topics", func(t *testing.T) { + sinkConfig := &config.SinkConfig{DispatchRules: []*config.DispatchRule{ + {Matcher: []string{"db1.t1"}, PartitionRule: "table", TopicRule: "topic-a"}, + {Matcher: []string{"db2.t2"}, PartitionRule: "table", TopicRule: "topic-b"}, + }} + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, sinkConfig) + kafkaSink.SetTableSchemaStore(commonEvent.NewTableSchemaStore([]*heartbeatpb.SchemaInfo{ + {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, + {SchemaName: "db2", Tables: []*heartbeatpb.TableInfo{{TableName: "t2"}}}, + }, common.KafkaSinkType, false)) + // The checkpoint must be fanned out to every active topic: the two + // rule topics and the default topic. + partitionCounts := map[string]int32{"topic-a": 2, "topic-b": 3, kafkaSinkTestTopic: 4} + for topic, partitionCount := range partitionCounts { + topicManager.EXPECT().GetPartitionNum(gomock.Any(), topic).Return(partitionCount, nil) + syncProducer.EXPECT().SendMessages(topic, partitionCount, gomock.Any()).Return(nil) + } + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("topic manager error", func(t *testing.T) { + kafkaSink, topicManager, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, &config.SinkConfig{}) + cause := context.DeadlineExceeded + topicManager.EXPECT().GetPartitionNum(gomock.Any(), kafkaSinkTestTopic).Return(int32(0), cause) + kafkaSink.checkpointChan <- 100 + + require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("producer error stops active topic fan-out", func(t *testing.T) { + sinkConfig := &config.SinkConfig{DispatchRules: []*config.DispatchRule{ + {Matcher: []string{"db1.t1"}, PartitionRule: "table", TopicRule: "topic-a"}, + }} + kafkaSink, topicManager, _, syncProducer := newKafkaSinkForTest( + t, t.Context(), config.ProtocolOpen, sinkConfig) + kafkaSink.SetTableSchemaStore(commonEvent.NewTableSchemaStore([]*heartbeatpb.SchemaInfo{ + {SchemaName: "db1", Tables: []*heartbeatpb.TableInfo{{TableName: "t1"}}}, + }, common.KafkaSinkType, false)) + cause := errors.ErrKafkaSendMessage.GenWithStackByArgs() + // Fail whichever topic the fan-out reaches first: sendCheckpoint must + // return the error and stop, so exactly one GetPartitionNum and one + // SendMessages call are expected regardless of the topic order. + topicManager.EXPECT().GetPartitionNum(gomock.Any(), gomock.Any()).Return(int32(2), nil) + syncProducer.EXPECT().SendMessages(gomock.Any(), int32(2), gomock.Any()).Return(cause) + kafkaSink.checkpointChan <- 100 + + require.Equal(t, cause, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("nil encoded message", func(t *testing.T) { + kafkaSink, _, _, _ := newKafkaSinkForTest( + t, t.Context(), config.ProtocolCanalJSON, &config.SinkConfig{}) + kafkaSink.checkpointChan <- 100 + close(kafkaSink.checkpointChan) + + require.NoError(t, kafkaSink.sendCheckpoint(t.Context())) + }) + + t.Run("context cancellation", func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + kafkaSink, _, _, _ := newKafkaSinkForTest(t, ctx, config.ProtocolOpen, &config.SinkConfig{}) + cause := errors.ErrKafkaSinkClosed.GenWithStackByArgs() + cancel(cause) + + require.Equal(t, cause, kafkaSink.sendCheckpoint(ctx)) + }) +} + +func newKafkaSinkForTest( + t *testing.T, ctx context.Context, protocol config.Protocol, sinkConfig *config.SinkConfig, +) (*sink, *topicmanager.MockTopicManager, *kafka.MockAsyncProducer, *kafka.MockSyncProducer) { + t.Helper() + + ctrl := gomock.NewController(t) + changefeedID := common.NewChangefeedID4Test("test", t.Name()) + protocolName := protocol.String() + testSinkConfig := *sinkConfig + testSinkConfig.Protocol = &protocolName + sinkConfig = &testSinkConfig + router, err := eventrouter.NewEventRouter(sinkConfig, kafkaSinkTestTopic, false, false) + require.NoError(t, err) + columnSelector, err := columnselector.New(sinkConfig) + require.NoError(t, err) + encoderConfig := codecCommon.NewConfig(protocol).WithChangefeedID(changefeedID) + encoderConfig.MaxBatchSize = 1 + encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) + require.NoError(t, err) + encoder, err := codec.NewEventEncoder(ctx, encoderConfig, nil) + require.NoError(t, err) + topicManager := topicmanager.NewMockTopicManager(ctrl) + asyncProducer := kafka.NewMockAsyncProducer(ctrl) + syncProducer := kafka.NewMockSyncProducer(ctrl) + topicManager.EXPECT().Close().AnyTimes() + asyncProducer.EXPECT().Close().AnyTimes() + syncProducer.EXPECT().Close().AnyTimes() + factory := kafka.NewMockFactory(ctrl) + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) + factory.EXPECT().MetricsCollector(nil).Return(noopMetricsCollector{}) + + kafkaSink, err := newWithComponents(ctx, changefeedID, common.DefaultKeyspaceID, protocol, components{ + encoderGroup: encoderGroup, + encoder: encoder, + columnSelector: columnSelector, + eventRouter: router, + topicManager: topicManager, + factory: factory, + }) + require.NoError(t, err) + t.Cleanup(kafkaSink.Close) + + return kafkaSink, topicManager, asyncProducer, syncProducer +} +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 291aef0596..91cac43e1b 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -236,6 +236,9 @@ func (m *kafkaTopicManager) waitUntilTopicVisible( }, retry.WithBackoffBaseDelay(500), retry.WithBackoffMaxDelay(1000), retry.WithMaxTries(6), + retry.WithIsRetryableErr(func(err error) bool { + return !kafka.IsUnretryableKafkaError(err) + }), ) return err @@ -273,6 +276,7 @@ func (m *kafkaTopicManager) createTopic( return 0, cerror.WrapError(cerror.ErrKafkaCreateTopic, err) } +<<<<<<< HEAD log.Info( "Kafka admin client create the topic success", zap.String("keyspace", m.changefeedID.Keyspace()), @@ -284,6 +288,8 @@ func (m *kafkaTopicManager) createTopic( ) m.tryUpdatePartitionsAndLogging(topicName, m.cfg.PartitionNum) +======= +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) return m.cfg.PartitionNum, nil } @@ -291,6 +297,7 @@ func (m *kafkaTopicManager) createTopic( func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( ctx context.Context, topicName string, ) (int32, error) { +<<<<<<< HEAD // 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. @@ -312,6 +319,17 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( } } else if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok { return numPartition, nil +======= + // If the topic is not in the cache, try to get its metadata. + topicDetails, err := m.admin.GetTopicsMeta([]string{topicName}, false) + if err == nil { + if numPartition, ok := m.tryStoreTopicMeta(topicName, topicDetails); ok { + return numPartition, nil + } + } + if kafka.IsAuthorizationFailed(err) { + return m.useConfiguredPartitionNum(topicName, err), nil +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) } partitionNum, err := m.createTopic(ctx, topicName) @@ -326,6 +344,7 @@ func (m *kafkaTopicManager) CreateTopicAndWaitUntilVisible( if err != nil { return 0, errors.Trace(err) } + m.tryUpdatePartitionsAndLogging(topicName, partitionNum) return partitionNum, nil } diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 9094b5758b..669b45b114 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -15,11 +15,19 @@ package topicmanager import ( "context" +<<<<<<< HEAD "sync" +======= + "io" +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) "testing" "time" "github.com/IBM/sarama" +<<<<<<< HEAD +======= + "github.com/golang/mock/gomock" +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/sink/kafka" "github.com/stretchr/testify/require" @@ -132,8 +140,209 @@ func TestCreateTopic(t *testing.T) { } changefeedID := common.NewChangefeedID4Test("test", "test") +<<<<<<< HEAD ctx := context.Background() manager := newKafkaTopicManager(ctx, kafka.DefaultMockTopicName, changefeedID, adminClient, cfg) +======= + + t.Run("existing topic", func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{kafkaTopicManagerTestTopic}, false).Return( + map[string]kafka.TopicDetail{ + kafkaTopicManagerTestTopic: {Name: kafkaTopicManagerTestTopic, NumPartitions: 2}, + }, nil) + manager := newKafkaTopicManager( + kafkaTopicManagerTestTopic, + changefeedID, + adminClient, + &kafka.AutoCreateTopicConfig{PartitionNum: 2}, + ) + + partitionNum, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), kafkaTopicManagerTestTopic) + + require.NoError(t, err) + require.Equal(t, int32(2), partitionNum) + }) + + t.Run("create missing topic", func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + var createdTopic *kafka.TopicDetail + postCreateDescribeCount := 0 + var manager *kafkaTopicManager + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).DoAndReturn( + func([]string, bool) (map[string]kafka.TopicDetail, error) { + if createdTopic == nil { + return nil, errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrUnknownTopicOrPartition, "describe-topic", "new-topic") + } + postCreateDescribeCount++ + _, cached := manager.topics.Load("new-topic") + require.False(t, cached) + if postCreateDescribeCount == 1 { + return nil, errors.WrapError(errors.ErrKafkaAdminAPI, io.EOF, "describe-topic", "new-topic") + } + return map[string]kafka.TopicDetail{ + createdTopic.Name: {Name: createdTopic.Name, NumPartitions: createdTopic.NumPartitions}, + }, nil + }).Times(3) + adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn( + func(detail *kafka.TopicDetail) error { + copy := *detail + createdTopic = © + return nil + }) + manager = newKafkaTopicManager( + kafkaTopicManagerTestTopic, + changefeedID, + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForLocal, + }, + ) + + partitionNum, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), "new-topic") + + require.NoError(t, err) + require.Equal(t, int32(2), partitionNum) + require.Equal(t, &kafka.TopicDetail{ + Name: "new-topic", + NumPartitions: 2, + ReplicationFactor: 1, + }, createdTopic) + require.Equal(t, 2, postCreateDescribeCount) + partitionsNum, err := manager.GetPartitionNum(context.Background(), "new-topic") + require.NoError(t, err) + require.Equal(t, int32(2), partitionsNum) + }) + + t.Run("auto create disabled", func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return(map[string]kafka.TopicDetail{}, nil) + manager := newKafkaTopicManager( + "new-topic", + changefeedID, + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: false, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + }, + ) + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), "new-topic") + + require.ErrorContains(t, err, "`auto-create-topic` is false, and new-topic not found") + }) + + t.Run("create error", func(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return(map[string]kafka.TopicDetail{}, nil) + var createdTopic *kafka.TopicDetail + adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn( + func(detail *kafka.TopicDetail) error { + copy := *detail + createdTopic = © + return errors.ErrKafkaAdminAPI.GenWithStackByArgs("create-topic", detail.Name) + }) + manager := newKafkaTopicManager( + "new-topic", + changefeedID, + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 4, + }, + ) + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), "new-topic") + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.Equal(t, "new-topic", createdTopic.Name) + }) +} + +func TestCreateTopicValidatesReplicationFactor(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"new-topic"}, false).Return(map[string]kafka.TopicDetail{}, nil) + adminClient.EXPECT().GetBrokerConfig(kafka.MinInsyncReplicasConfigName).Return("2", true, nil) + manager := newKafkaTopicManager( + "new-topic", + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + }, + ) + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), "new-topic") + + require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker") +} + +func TestWaitUntilTopicVisibleUnretryableError(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"invalid-topic"}, false).Return( + nil, + errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidTopic, "describe-topic", "invalid-topic"), + ).Times(1) + manager := newKafkaTopicManager( + "invalid-topic", + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{PartitionNum: 2}, + ) + + err := manager.waitUntilTopicVisible(context.Background(), "invalid-topic") + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrInvalidTopic) +} + +func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"existing-topic"}, false).Return( + map[string]kafka.TopicDetail{ + "existing-topic": {Name: "existing-topic", NumPartitions: 2}, + }, nil) + + manager, err := GetTopicManagerAndTryCreateTopic( + t.Context(), + common.NewChangefeedID4Test("test", "test"), + "existing-topic", + &kafka.AutoCreateTopicConfig{PartitionNum: 2}, + adminClient, + ) + + require.NoError(t, err) +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) defer manager.Close() partitionNum, err := manager.CreateTopicAndWaitUntilVisible(ctx, kafka.DefaultMockTopicName) require.NoError(t, err) @@ -178,6 +387,7 @@ func TestCreateTopic(t *testing.T) { func TestCreateTopicWithTopicDescribeDenied(t *testing.T) { t.Parallel() +<<<<<<< HEAD adminClient := &mockAdminClientWithDeniedDescribe{ ClusterAdminClientMockImpl: kafka.NewClusterAdminClientMockImpl(), } @@ -187,6 +397,22 @@ func TestCreateTopicWithTopicDescribeDenied(t *testing.T) { PartitionNum: 2, ReplicationFactor: 1, } +======= + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"default-topic"}, false).Return( + nil, errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs("describe-topic", "default-topic")) + manager := newKafkaTopicManager( + "default-topic", + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + }, + ) +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) changefeedID := common.NewChangefeedID4Test("test", "test") ctx := context.Background() @@ -207,6 +433,7 @@ func TestCreateTopicWithTopicDescribeDenied(t *testing.T) { func TestCreateTopicWithCreateDenied(t *testing.T) { t.Parallel() +<<<<<<< HEAD adminClient := &mockAdminClientWithDeniedCreate{ ClusterAdminClientMockImpl: kafka.NewClusterAdminClientMockImpl(), } @@ -214,6 +441,14 @@ func TestCreateTopicWithCreateDenied(t *testing.T) { cfg := &kafka.AutoCreateTopicConfig{ AutoCreate: true, PartitionNum: 2, +======= + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{"default-topic"}, false).Return(map[string]kafka.TopicDetail{}, nil) + adminClient.EXPECT().CreateTopic(&kafka.TopicDetail{ + Name: "default-topic", + NumPartitions: 2, +>>>>>>> 51db5185d (kafka: improve stability when creating many topics with Kafka v4 (#6081)) ReplicationFactor: 1, } diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 4141b6a112..2da1a4dbdb 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -132,9 +132,6 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool for _, meta := range metaList { if meta.Err != sarama.ErrNoError { - if meta.Err == sarama.ErrUnknownTopicOrPartition { - continue - } if !ignoreTopicError { return nil, meta.Err } @@ -159,6 +156,26 @@ func IsAdminAuthorizationFailed(err error) bool { errors.Is(err, sarama.ErrClusterAuthorizationFailed) } +// IsUnretryableKafkaError reports whether err is not retryable. +// See Apache Kafka protocol error definitions: +// https://kafka.apache.org/38/generated/protocol_errors.html +func IsUnretryableKafkaError(err error) bool { + if IsAuthorizationFailed(err) || + errors.Is(err, errors.ErrKafkaInvalidConfig) || + errors.Is(err, sarama.ErrInvalidTopic) || + errors.Is(err, sarama.ErrInvalidConfig) || + errors.Is(err, sarama.ErrSASLAuthenticationFailed) || + errors.Is(err, sarama.ErrUnsupportedSASLMechanism) || + errors.Is(err, sarama.ErrIllegalSASLState) || + errors.Is(err, sarama.ErrUnsupportedVersion) || + errors.Is(err, sarama.ErrInvalidRequest) { + return true + } + + var configErr sarama.ConfigurationError + return errors.As(err, &configErr) +} + 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 84abedae7e..e93108329f 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -578,6 +578,8 @@ func adjustOptions( options *options, topic string, ) error { + // The topic may not exist yet and will be created later by the topic manager, + // so ignore per-topic metadata errors here. topics, err := admin.GetTopicsMeta([]string{topic}, true) if err != nil { return errors.Trace(err) diff --git a/pkg/sink/kafka/sarama_admin_test.go b/pkg/sink/kafka/sarama_admin_test.go new file mode 100644 index 0000000000..d0ce07fc1d --- /dev/null +++ b/pkg/sink/kafka/sarama_admin_test.go @@ -0,0 +1,449 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "io" + "testing" + + "github.com/IBM/sarama" + "github.com/golang/mock/gomock" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/stretchr/testify/require" +) + +func TestGetBrokerConfig(t *testing.T) { + t.Parallel() + + t.Run("found", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeCluster().Return(nil, int32(1), nil) + admin.EXPECT().DescribeConfig(sarama.ConfigResource{ + Type: sarama.BrokerResource, + Name: "1", + ConfigNames: []string{"message.max.bytes"}, + }).Return([]sarama.ConfigEntry{ + {Name: "unrelated", Value: "value"}, + {Name: "message.max.bytes", Value: "1048576"}, + }, nil) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + value, found, err := client.GetBrokerConfig("message.max.bytes") + + require.NoError(t, err) + require.True(t, found) + require.Equal(t, "1048576", value) + }) + + t.Run("not found", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeCluster().Return(nil, int32(1), nil) + admin.EXPECT().DescribeConfig(gomock.Any()).Return([]sarama.ConfigEntry{}, nil) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + value, found, err := client.GetBrokerConfig("missing") + + require.NoError(t, err) + require.False(t, found) + require.Empty(t, value) + }) + + t.Run("admin error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + cause := io.ErrUnexpectedEOF + admin.EXPECT().DescribeCluster().Return(nil, int32(0), cause) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + _, _, err := client.GetBrokerConfig("missing") + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, cause) + }) +} + +func TestGetTopicConfig(t *testing.T) { + t.Parallel() + + t.Run("found", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeConfig(sarama.ConfigResource{ + Type: sarama.TopicResource, + Name: "test-topic", + ConfigNames: []string{"max.message.bytes"}, + }).Return([]sarama.ConfigEntry{ + {Name: "max.message.bytes", Value: "1048576"}, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + value, found, err := client.GetTopicConfig("test-topic", "max.message.bytes") + + require.NoError(t, err) + require.True(t, found) + require.Equal(t, "1048576", value) + }) + + t.Run("not found", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeConfig(gomock.Any()).Return([]sarama.ConfigEntry{}, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + value, found, err := client.GetTopicConfig("test-topic", "missing") + + require.NoError(t, err) + require.False(t, found) + require.Empty(t, value) + }) + + t.Run("admin error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeConfig(gomock.Any()).Return(nil, context.DeadlineExceeded) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + _, _, err := client.GetTopicConfig("test-topic", "missing") + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.False(t, IsAuthorizationFailed(err)) + }) +} + +func TestGetTopicsMeta(t *testing.T) { + t.Parallel() + + t.Run("returns unknown topic error when topic errors are not ignored", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"valid-topic", "missing-topic"}).Return([]*sarama.TopicMetadata{ + { + Name: "valid-topic", + Partitions: []*sarama.PartitionMetadata{{}, {}}, + }, + { + Name: "missing-topic", + Err: sarama.ErrUnknownTopicOrPartition, + }, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + topics, err := client.GetTopicsMeta([]string{"valid-topic", "missing-topic"}, false) + + require.Nil(t, topics) + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrUnknownTopicOrPartition) + }) + + t.Run("ignores unknown topic error and returns valid topics", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"valid-topic", "missing-topic"}).Return([]*sarama.TopicMetadata{ + { + Name: "valid-topic", + Partitions: []*sarama.PartitionMetadata{{}, {}}, + }, + { + Name: "missing-topic", + Err: sarama.ErrUnknownTopicOrPartition, + }, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + topics, err := client.GetTopicsMeta([]string{"valid-topic", "missing-topic"}, true) + + require.NoError(t, err) + require.Equal(t, map[string]TopicDetail{ + "valid-topic": { + Name: "valid-topic", + NumPartitions: 2, + }, + }, topics) + }) + + t.Run("missing response", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"missing-topic"}).Return(nil, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + topics, err := client.GetTopicsMeta([]string{"missing-topic"}, false) + + require.NoError(t, err) + require.Empty(t, topics) + }) + + t.Run("topic error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ + {Name: "test-topic", Err: sarama.ErrInvalidTopic}, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + _, err := client.GetTopicsMeta([]string{"test-topic"}, false) + + require.ErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrInvalidTopic) + require.False(t, IsAuthorizationFailed(err)) + }) + + t.Run("topic authorization error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ + {Name: "test-topic", Err: sarama.ErrTopicAuthorizationFailed}, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + _, err := client.GetTopicsMeta([]string{"test-topic"}, false) + + require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) + require.NotErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrTopicAuthorizationFailed) + require.True(t, IsAuthorizationFailed(err)) + code, ok := errors.RFCCode(err) + require.True(t, ok) + require.Equal(t, errors.ErrKafkaAuthorizationFailed.RFCCode(), code) + }) + + t.Run("cluster authorization error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return(nil, sarama.ErrClusterAuthorizationFailed) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + _, err := client.GetTopicsMeta([]string{"test-topic"}, false) + + require.ErrorIs(t, err, errors.ErrKafkaAuthorizationFailed) + require.NotErrorIs(t, err, errors.ErrKafkaAdminAPI) + require.ErrorIs(t, err, sarama.ErrClusterAuthorizationFailed) + require.True(t, IsAuthorizationFailed(err)) + }) + + t.Run("ignores non-unknown topic error", func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ + {Name: "test-topic", Err: sarama.ErrInvalidTopic}, + }, nil) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + topics, err := client.GetTopicsMeta([]string{"test-topic"}, true) + + require.NoError(t, err) + require.Empty(t, topics) + }) +} + +func TestIsAuthorizationFailed(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + expected bool + }{ + {name: "TiCDC authorization error", err: errors.ErrKafkaAuthorizationFailed.GenWithStackByArgs("describe-topic", "test-topic"), expected: true}, + {name: "topic authorization error", err: sarama.ErrTopicAuthorizationFailed, expected: true}, + {name: "cluster authorization error", err: sarama.ErrClusterAuthorizationFailed, expected: true}, + {name: "general error", err: sarama.ErrInvalidTopic}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.expected, IsAuthorizationFailed(test.err)) + }) + } +} + +func TestIsUnretryableKafkaError(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + unretryable bool + }{ + {name: "unknown topic", err: sarama.ErrUnknownTopicOrPartition}, + {name: "leader unavailable", err: sarama.ErrLeaderNotAvailable}, + {name: "request timeout", err: sarama.ErrRequestTimedOut}, + {name: "network exception", err: sarama.ErrNetworkException}, + {name: "controller changed", err: sarama.ErrNotController}, + {name: "no broker available", err: sarama.ErrOutOfBrokers}, + {name: "EOF", err: io.EOF}, + {name: "unknown broker error", err: sarama.ErrUnknown}, + { + name: "wrapped unknown topic", + err: errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrUnknownTopicOrPartition, "describe-topic", "test-topic"), + }, + {name: "context cancellation", err: context.Canceled}, + {name: "TiCDC invalid config", err: errors.ErrKafkaInvalidConfig.GenWithStack("invalid config"), unretryable: true}, + {name: "topic authorization failure", err: sarama.ErrTopicAuthorizationFailed, unretryable: true}, + {name: "cluster authorization failure", err: sarama.ErrClusterAuthorizationFailed, unretryable: true}, + {name: "invalid topic", err: sarama.ErrInvalidTopic, unretryable: true}, + {name: "invalid config", err: sarama.ErrInvalidConfig, unretryable: true}, + {name: "SASL authentication failure", err: sarama.ErrSASLAuthenticationFailed, unretryable: true}, + {name: "unsupported SASL mechanism", err: sarama.ErrUnsupportedSASLMechanism, unretryable: true}, + {name: "illegal SASL state", err: sarama.ErrIllegalSASLState, unretryable: true}, + {name: "unsupported version", err: sarama.ErrUnsupportedVersion, unretryable: true}, + {name: "invalid request", err: sarama.ErrInvalidRequest, unretryable: true}, + {name: "client configuration error", err: sarama.ConfigurationError("invalid client config"), unretryable: true}, + {name: "wrapped client configuration error", err: errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ConfigurationError("invalid client config"), "describe-topic", "test-topic"), unretryable: true}, + { + name: "wrapped invalid topic", + err: errors.WrapError(errors.ErrKafkaAdminAPI, sarama.ErrInvalidTopic, "describe-topic", "test-topic"), + unretryable: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.unretryable, IsUnretryableKafkaError(test.err)) + }) + } +} + +func TestCreateTopic(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + adminErr error + expectedErr error + authorization bool + }{ + {name: "success"}, + {name: "topic already exists", adminErr: sarama.ErrTopicAlreadyExists}, + {name: "authorization error", adminErr: sarama.ErrClusterAuthorizationFailed, expectedErr: errors.ErrKafkaAuthorizationFailed, authorization: true}, + {name: "general error", adminErr: sarama.ErrInvalidReplicationFactor, expectedErr: errors.ErrKafkaAdminAPI}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().CreateTopic("test-topic", &sarama.TopicDetail{ + NumPartitions: 3, + ReplicationFactor: 2, + }, false).Return(test.adminErr) + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + + err := client.CreateTopic(&TopicDetail{ + Name: "test-topic", + NumPartitions: 3, + ReplicationFactor: 2, + }) + + if test.expectedErr == nil { + require.NoError(t, err) + return + } + require.ErrorIs(t, err, test.expectedErr) + require.ErrorIs(t, err, test.adminErr) + if test.authorization { + require.NotErrorIs(t, err, errors.ErrKafkaAdminAPI) + } + }) + } +} + +func TestAdminClientClose(t *testing.T) { + tests := []struct { + name string + setup func(*gomock.Controller) *saramaAdminClient + }{ + { + name: "uses admin close", + setup: func(ctrl *gomock.Controller) *saramaAdminClient { + client := NewMocksaramaClient(ctrl) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().Close().Return(nil) + client.EXPECT().Close().Times(0) + return &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + client: client, + admin: admin, + } + }, + }, + { + name: "falls back to client when admin is nil", + setup: func(ctrl *gomock.Controller) *saramaAdminClient { + client := NewMocksaramaClient(ctrl) + client.EXPECT().Close().Return(nil) + return &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + client: client, + } + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := test.setup(ctrl) + + require.NotPanics(t, func() { adminClient.Close() }) + }) + } +}