From 980e28cdf45ac8b138e6057f7ee92404c5b1d7cb Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 23 Jul 2026 19:29:36 +0800 Subject: [PATCH 1/9] enhance the verify kafka sink --- downstreamadapter/sink/kafka/helper.go | 26 ++++----- downstreamadapter/sink/kafka/sink.go | 57 ++++++++++++------- downstreamadapter/sink/kafka/sink_test.go | 27 +++++++++ pkg/sink/codec/canal/canal_json_encoder.go | 2 +- pkg/sink/codec/encoder_group.go | 45 ++++++++------- pkg/sink/codec/open/encoder.go | 2 +- pkg/sink/codec/simple/encoder.go | 2 +- pkg/sink/kafka/admin.go | 18 +++++- pkg/sink/kafka/admin_test.go | 24 ++++++++ pkg/sink/kafka/claimcheck/claim_check.go | 6 ++ pkg/sink/kafka/claimcheck/claim_check_test.go | 14 +++++ pkg/sink/kafka/options.go | 8 ++- pkg/sink/kafka/options_test.go | 20 +++++-- 13 files changed, 185 insertions(+), 66 deletions(-) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index 6665b4c973..194ff42850 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -41,6 +41,9 @@ type components struct { } func (c components) close() { + if c.encoder != nil { + c.encoder.Clean() + } if c.adminClient != nil { c.adminClient.Close() } @@ -94,29 +97,21 @@ func newKafkaSinkComponent( return kafkaComponent, protocol, errors.Trace(err) } - kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) - if err != nil { - return kafkaComponent, protocol, errors.Trace(err) - } - kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig) if err != nil { return kafkaComponent, protocol, errors.Trace(err) } + defer func() { + if err != nil { + kafkaComponent.close() + } + }() kafkaComponent.adminClient, err = kafkaComponent.factory.AdminClient(ctx) if err != nil { return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) } - // We must close adminClient when this func return cause by an error - // otherwise the adminClient will never be closed and lead to a goroutine leak. - defer func() { - if err != nil && kafkaComponent.adminClient != nil { - kafkaComponent.adminClient.Close() - } - }() - kafkaComponent.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( ctx, changefeedID, @@ -127,5 +122,10 @@ func newKafkaSinkComponent( if err != nil { return kafkaComponent, protocol, errors.Trace(err) } + + kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) + if err != nil { + return kafkaComponent, protocol, errors.Trace(err) + } return kafkaComponent, protocol, nil } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 673d3e9993..291d39265f 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -87,11 +87,6 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. } options.Topic = topic - encoderConfig, err := helper.GetEncoderConfig(changefeedID, uri, protocol, sinkConfig, options.MaxMessageBytes) - if err != nil { - return errors.Trace(err) - } - isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { return errors.Trace(err) @@ -106,33 +101,52 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.WrapError(errors.ErrKafkaNewProducer, err) } + encoderConfig, err := helper.GetEncoderConfig( + commonType.NewChangefeedID(changefeedID.Keyspace()), + uri, + protocol, + sinkConfig, + options.MaxMessageBytes, + ) + if err != nil { + return errors.Trace(err) + } + adminClient, err := factory.AdminClient(ctx) if err != nil { return errors.WrapError(errors.ErrKafkaNewProducer, err) } defer adminClient.Close() + return verifyKafkaSink(ctx, adminClient, topic, options.DeriveTopicConfig(), encoderConfig) +} + +func verifyKafkaSink( + ctx context.Context, + adminClient kafka.ClusterAdminClient, + topic string, + topicConfig *kafka.AutoCreateTopicConfig, + encoderConfig *common.Config, +) error { topics, err := adminClient.GetTopicsMeta([]string{topic}, false) if err != nil { return errors.Trace(err) } - if _, exists := topics[topic]; exists { - return nil - } - - topicConfig := options.DeriveTopicConfig() - if !topicConfig.AutoCreate { - return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) - } + if _, exists := topics[topic]; !exists { + if !topicConfig.AutoCreate { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "`auto-create-topic` is false, and %s not found", topic) + } - // the topic is not created, only validate. - err = adminClient.CreateTopic(&kafka.TopicDetail{ - Name: topic, - NumPartitions: topicConfig.PartitionNum, - ReplicationFactor: topicConfig.ReplicationFactor, - }, true) - if err != nil { - return errors.WrapError(errors.ErrKafkaCreateTopic, err) + // The topic is not created, only validate its configuration and permissions. + err = adminClient.CreateTopic(&kafka.TopicDetail{ + Name: topic, + NumPartitions: topicConfig.PartitionNum, + ReplicationFactor: topicConfig.ReplicationFactor, + }, true) + if err != nil { + return errors.WrapError(errors.ErrKafkaCreateTopic, err) + } } encoder, err := codec.NewEventEncoder(ctx, encoderConfig) @@ -140,7 +154,6 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.Trace(err) } encoder.Clean() - return nil } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 440806501c..ca3b82a328 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -244,3 +244,30 @@ func TestKafkaSinkBatchConfig(t *testing.T) { require.Equal(t, 4096, sink.BatchCount()) require.Zero(t, sink.BatchBytes()) } + +func TestVerifyKafkaSinkRejectsInvalidClaimCheckStorageForExistingTopic(t *testing.T) { + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, false).Return( + map[string]kafka.TopicDetail{ + kafkaSinkTestTopic: { + Name: kafkaSinkTestTopic, + NumPartitions: 1, + ReplicationFactor: 1, + }, + }, nil) + + encoderConfig := codeccommon.NewConfig(config.ProtocolOpen) + encoderConfig.ChangefeedID = common.NewChangefeedID4Test("test", "test") + encoderConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck + encoderConfig.LargeMessageHandle.ClaimCheckStorageURI = "unsupported:///claim-check" + + err := verifyKafkaSink( + context.Background(), + adminClient, + kafkaSinkTestTopic, + &kafka.AutoCreateTopicConfig{AutoCreate: false}, + encoderConfig, + ) + require.Error(t, err) +} diff --git a/pkg/sink/codec/canal/canal_json_encoder.go b/pkg/sink/codec/canal/canal_json_encoder.go index 7425a666ef..d144d62a46 100644 --- a/pkg/sink/codec/canal/canal_json_encoder.go +++ b/pkg/sink/codec/canal/canal_json_encoder.go @@ -585,6 +585,6 @@ func (c *JSONRowEventEncoder) EncodeDDLEvent(e *commonEvent.DDLEvent) (*common.M func (c *JSONRowEventEncoder) Clean() { if c.claimCheck != nil { - c.claimCheck.CleanMetrics() + c.claimCheck.Close() } } diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go index 7ae503c985..d7d781f53e 100644 --- a/pkg/sink/codec/encoder_group.go +++ b/pkg/sink/codec/encoder_group.go @@ -73,29 +73,39 @@ func NewEncoderGroup( if concurrency <= 0 { concurrency = config.DefaultEncoderGroupConcurrency } - inputCh := make([]chan *future, concurrency) - rowEventEncoders := make([]common.EventEncoder, concurrency) + group := &encoderGroup{ + changefeedID: changefeedID, + rowEventEncoders: make([]common.EventEncoder, concurrency), + concurrency: concurrency, + inputCh: make([]chan *future, concurrency), + outputCh: make(chan *future, defaultInputChanSize*concurrency), + } + initialized := false + defer func() { + if !initialized { + group.close() + } + }() + var err error for i := 0; i < concurrency; i++ { - inputCh[i] = make(chan *future, defaultInputChanSize) - rowEventEncoders[i], err = NewEventEncoder(ctx, encoderConfig) + group.inputCh[i] = make(chan *future, defaultInputChanSize) + group.rowEventEncoders[i], err = NewEventEncoder(ctx, encoderConfig) if err != nil { log.Error("failed to create row event encoder", zap.Error(err)) return nil, errors.Trace(err) } } - outCh := make(chan *future, defaultInputChanSize*concurrency) - var bw *bootstrapWorker if cfg.ShouldSendBootstrapMsg() { encoder, err := NewEventEncoder(ctx, encoderConfig) if err != nil { log.Error("failed to create row event encoder", zap.Error(err)) return nil, errors.Trace(err) } - bw = newBootstrapWorker( + group.bootstrapWorker = newBootstrapWorker( changefeedID, - outCh, + group.outputCh, encoder, util.GetOrZero(cfg.SendBootstrapIntervalInSec), util.GetOrZero(cfg.SendBootstrapInMsgCount), @@ -104,20 +114,13 @@ func NewEncoderGroup( ) } - return &encoderGroup{ - changefeedID: changefeedID, - rowEventEncoders: rowEventEncoders, - concurrency: concurrency, - inputCh: inputCh, - index: 0, - outputCh: outCh, - bootstrapWorker: bw, - }, nil + initialized = true + return group, nil } func (g *encoderGroup) Run(ctx context.Context) error { defer func() { - g.cleanMetrics() + g.close() log.Info("encoder group exited", zap.String("keyspace", g.changefeedID.Keyspace()), zap.String("changefeed", g.changefeedID.Name())) @@ -204,10 +207,12 @@ func (g *encoderGroup) Output() <-chan *future { return g.outputCh } -func (g *encoderGroup) cleanMetrics() { +func (g *encoderGroup) close() { encoderGroupInputChanSizeGauge.DeleteLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name()) for _, encoder := range g.rowEventEncoders { - encoder.Clean() + if encoder != nil { + encoder.Clean() + } } common.CleanMetrics(g.changefeedID) } diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index 75f82e9d03..1f0db5f6ce 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -66,7 +66,7 @@ func NewBatchEncoder(ctx context.Context, config *common.Config) (common.EventEn func (d *batchEncoder) Clean() { if d.claimCheck != nil { - d.claimCheck.CleanMetrics() + d.claimCheck.Close() } } diff --git a/pkg/sink/codec/simple/encoder.go b/pkg/sink/codec/simple/encoder.go index b8ef228561..9479478ae0 100644 --- a/pkg/sink/codec/simple/encoder.go +++ b/pkg/sink/codec/simple/encoder.go @@ -165,6 +165,6 @@ func (e *Encoder) EncodeDDLEvent(event *commonEvent.DDLEvent) (*common.Message, // CleanMetrics implement the RowEventEncoderBuilder interface func (e *Encoder) Clean() { if e.claimCheck != nil { - e.claimCheck.CleanMetrics() + e.claimCheck.Close() } } diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 16087c889c..0c58d8933f 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -146,13 +146,27 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool continue } result[meta.Name] = TopicDetail{ - Name: meta.Name, - NumPartitions: int32(len(meta.Partitions)), + Name: meta.Name, + NumPartitions: int32(len(meta.Partitions)), + ReplicationFactor: minReplicationFactor(meta.Partitions), } } return result, nil } +func minReplicationFactor(partitions []*sarama.PartitionMetadata) int16 { + minReplicas := 0 + for i, partition := range partitions { + if partition == nil { + return 0 + } + if i == 0 || len(partition.Replicas) < minReplicas { + minReplicas = len(partition.Replicas) + } + } + return int16(minReplicas) +} + // IsAdminAuthorizationFailed checks whether err is an authorization failure from Kafka admin APIs. func IsAdminAuthorizationFailed(err error) bool { return errors.Is(err, sarama.ErrTopicAuthorizationFailed) || diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index c2e3f90e37..67e20e565c 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -16,6 +16,7 @@ package kafka import ( "testing" + "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" "github.com/stretchr/testify/require" @@ -62,3 +63,26 @@ func TestAdminClientClose(t *testing.T) { }) } } + +func TestGetTopicsMetaUsesSmallestReplicationFactor(t *testing.T) { + ctrl := gomock.NewController(t) + admin := NewMocksaramaClusterAdmin(ctrl) + admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ + { + Name: "test-topic", + Err: sarama.ErrNoError, + Partitions: []*sarama.PartitionMetadata{ + {Replicas: []int32{1, 2, 3}}, + {Replicas: []int32{1, 2}}, + }, + }, + }, nil) + + client := &saramaAdminClient{ + changefeed: common.NewChangeFeedIDWithName("test", "default"), + admin: admin, + } + topics, err := client.GetTopicsMeta([]string{"test-topic"}, false) + require.NoError(t, err) + require.Equal(t, int16(2), topics["test-topic"].ReplicationFactor) +} diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index 1b25528be2..fea8ba6cc9 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -114,6 +114,12 @@ func (c *ClaimCheck) CleanMetrics() { claimCheckSendMessageCount.DeleteLabelValues(c.changefeedID.Keyspace(), c.changefeedID.Name()) } +// Close releases the external storage and removes the claim-check metrics. +func (c *ClaimCheck) Close() { + c.storage.Close() + c.CleanMetrics() +} + // NewFileName return the file name for the message which is delivered to the external storage system. // UUID V4 is used to generate random and unique file names. // This should not exceed the S3 object name length limit. diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go index a641ddd97b..e1f14ca877 100644 --- a/pkg/sink/kafka/claimcheck/claim_check_test.go +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -19,7 +19,9 @@ import ( commonType "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/tidb/pkg/objstore/mockobjstore" "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" ) func TestClaimCheck(t *testing.T) { @@ -42,3 +44,15 @@ func TestClaimCheck(t *testing.T) { fileName := claimCheck.FileNameWithPrefix("file.json") require.Equal(t, "file:///tmp/abc/file.json", fileName) } + +func TestClaimCheckCloseClosesStorage(t *testing.T) { + ctrl := gomock.NewController(t) + storage := mockobjstore.NewMockStorage(ctrl) + storage.EXPECT().Close().Times(1) + claimCheck := &ClaimCheck{ + storage: storage, + changefeedID: commonType.NewChangeFeedIDWithName("test", "default"), + } + + claimCheck.Close() +} diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 7c51166a9e..959a4820a9 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -586,18 +586,22 @@ func adjustOptions( if err != nil { return errors.Trace(err) } + info, exists := topics[topic] // Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1. // If we don't check it, the producer probably can not send message to the topic. // Because it will wait for the ack from all replicas. But we do not have enough replicas. if options.RequiredAcks == WaitForAll { - err = validateMinInsyncReplicas(ctx, admin, topics, topic, int(options.ReplicationFactor)) + replicationFactor := options.ReplicationFactor + if exists { + replicationFactor = info.ReplicationFactor + } + err = validateMinInsyncReplicas(ctx, admin, topics, topic, int(replicationFactor)) if err != nil { return errors.Trace(err) } } - info, exists := topics[topic] // once we have found the topic, no matter `auto-create-topic`, // make sure user input parameters are valid. if exists { diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 0650d3baf2..321683d424 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -84,7 +84,11 @@ func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { } func (f *kafkaAdminFixture) addTopic(name string, partitionNum int32) { - f.topics[name] = TopicDetail{Name: name, NumPartitions: partitionNum} + f.topics[name] = TopicDetail{ + Name: name, + NumPartitions: partitionNum, + ReplicationFactor: mockClusterReplicationFactor, + } } func (f *kafkaAdminFixture) getTopicsMeta( @@ -402,8 +406,9 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t adminClient := adminFixture.admin detail := &TopicDetail{ - Name: topicName, - NumPartitions: 3, + Name: topicName, + NumPartitions: 3, + ReplicationFactor: mockClusterReplicationFactor, } err := adminClient.CreateTopic(detail, false) require.NoError(t, err) @@ -478,9 +483,16 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) { err = adjustOptions(ctx, adminClient, options, topicName) require.Nil(t, err) - // topic found, and have `min.insync.replicas`, but set to 2, larger than `replication-factor`. + // Existing topics use their actual replication factor rather than the option + // used only when creating a topic. adminFixture.setMinInsyncReplicas("2") err = adjustOptions(ctx, adminClient, options, defaultMockTopicName) + require.NoError(t, err) + + topicDetail := adminFixture.topics[defaultMockTopicName] + topicDetail.ReplicationFactor = 1 + adminFixture.topics[defaultMockTopicName] = topicDetail + err = adjustOptions(ctx, adminClient, options, defaultMockTopicName) require.Regexp(t, ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of topic.*", errors.Cause(err), From 3260a4748878f3bf77e4685bfed30a5bfd842a70 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 14:37:05 +0800 Subject: [PATCH 2/9] simplify the code --- pkg/sink/codec/builder.go | 13 +++++-- pkg/sink/codec/canal/canal_json_encoder.go | 29 +++++++++++----- pkg/sink/codec/encoder_group.go | 31 +++++++++++++++-- pkg/sink/codec/open/encoder.go | 29 +++++++++++----- pkg/sink/codec/simple/encoder.go | 32 ++++++++++++----- pkg/sink/kafka/claimcheck/claim_check.go | 3 +- pkg/sink/kafka/claimcheck/claim_check_test.go | 34 +++++++++++++++++++ 7 files changed, 141 insertions(+), 30 deletions(-) diff --git a/pkg/sink/codec/builder.go b/pkg/sink/codec/builder.go index e6415d06f6..c9e31d565b 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -27,23 +27,30 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec/debezium" "github.com/pingcap/ticdc/pkg/sink/codec/open" "github.com/pingcap/ticdc/pkg/sink/codec/simple" + "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" "go.uber.org/zap" ) func NewEventEncoder(ctx context.Context, cfg *common.Config) (common.EventEncoder, error) { + return newEventEncoder(ctx, cfg) +} + +func newEventEncoder( + ctx context.Context, cfg *common.Config, claimChecks ...*claimcheck.ClaimCheck, +) (common.EventEncoder, error) { switch cfg.Protocol { case config.ProtocolDefault, config.ProtocolOpen: - return open.NewBatchEncoder(ctx, cfg) + return open.NewBatchEncoder(ctx, cfg, claimChecks...) case config.ProtocolAvro: return avro.NewAvroEncoder(ctx, cfg) case config.ProtocolCanalJSON: - return canal.NewJSONRowEventEncoder(ctx, cfg) + return canal.NewJSONRowEventEncoder(ctx, cfg, claimChecks...) case config.ProtocolDebezium: return debezium.NewBatchEncoder(cfg, config.GetGlobalServerConfig().ClusterID), nil case config.ProtocolDebeziumAvro: return debezium.NewAvroBatchEncoder(ctx, cfg, config.GetGlobalServerConfig().ClusterID) case config.ProtocolSimple: - return simple.NewEncoder(ctx, cfg) + return simple.NewEncoder(ctx, cfg, claimChecks...) default: return nil, errors.ErrSinkUnknownProtocol.GenWithStackByArgs(cfg.Protocol) } diff --git a/pkg/sink/codec/canal/canal_json_encoder.go b/pkg/sink/codec/canal/canal_json_encoder.go index d144d62a46..ca81b0a1a2 100644 --- a/pkg/sink/codec/canal/canal_json_encoder.go +++ b/pkg/sink/codec/canal/canal_json_encoder.go @@ -369,20 +369,33 @@ type JSONRowEventEncoder struct { messages []*common.Message claimCheck *claimcheck.ClaimCheck + cleanup func() config *common.Config } // NewJSONRowEventEncoder creates a new JSONRowEventEncoder -func NewJSONRowEventEncoder(ctx context.Context, config *common.Config) (common.EventEncoder, error) { - claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) - if err != nil { - return nil, err +func NewJSONRowEventEncoder( + ctx context.Context, config *common.Config, claimChecks ...*claimcheck.ClaimCheck, +) (common.EventEncoder, error) { + var claimCheck *claimcheck.ClaimCheck + if len(claimChecks) == 0 { + var err error + claimCheck, err = claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) + if err != nil { + return nil, err + } + } else { + claimCheck = claimChecks[0] } - return &JSONRowEventEncoder{ + encoder := &JSONRowEventEncoder{ messages: make([]*common.Message, 0, 1), config: config, claimCheck: claimCheck, - }, nil + } + if len(claimChecks) == 0 && claimCheck != nil { + encoder.cleanup = claimCheck.Close + } + return encoder, nil } func (c *JSONRowEventEncoder) newJSONMessageForDDL(e *commonEvent.DDLEvent) canalJSONMessageInterface { @@ -584,7 +597,7 @@ func (c *JSONRowEventEncoder) EncodeDDLEvent(e *commonEvent.DDLEvent) (*common.M } func (c *JSONRowEventEncoder) Clean() { - if c.claimCheck != nil { - c.claimCheck.Close() + if c.cleanup != nil { + c.cleanup() } } diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go index d7d781f53e..124e2253c8 100644 --- a/pkg/sink/codec/encoder_group.go +++ b/pkg/sink/codec/encoder_group.go @@ -25,6 +25,7 @@ import ( "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec/common" + "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/ticdc/pkg/util" "go.uber.org/zap" "golang.org/x/sync/errgroup" @@ -60,6 +61,7 @@ type encoderGroup struct { outputCh chan *future bootstrapWorker *bootstrapWorker + claimCheck *claimcheck.ClaimCheck } // NewEncoderGroup creates a new EncoderGroup instance @@ -68,6 +70,28 @@ func NewEncoderGroup( cfg *config.SinkConfig, encoderConfig *common.Config, changefeedID commonType.ChangeFeedID, +) (*encoderGroup, error) { + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return nil, errors.Trace(err) + } + group, err := newEncoderGroup(ctx, cfg, encoderConfig, changefeedID, claimCheck) + if err != nil { + if claimCheck != nil { + claimCheck.Close() + } + return nil, errors.Trace(err) + } + group.claimCheck = claimCheck + return group, nil +} + +func newEncoderGroup( + ctx context.Context, + cfg *config.SinkConfig, + encoderConfig *common.Config, + changefeedID commonType.ChangeFeedID, + claimCheck *claimcheck.ClaimCheck, ) (*encoderGroup, error) { concurrency := util.GetOrZero(cfg.EncoderConcurrency) if concurrency <= 0 { @@ -90,7 +114,7 @@ func NewEncoderGroup( var err error for i := 0; i < concurrency; i++ { group.inputCh[i] = make(chan *future, defaultInputChanSize) - group.rowEventEncoders[i], err = NewEventEncoder(ctx, encoderConfig) + group.rowEventEncoders[i], err = newEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { log.Error("failed to create row event encoder", zap.Error(err)) return nil, errors.Trace(err) @@ -98,7 +122,7 @@ func NewEncoderGroup( } if cfg.ShouldSendBootstrapMsg() { - encoder, err := NewEventEncoder(ctx, encoderConfig) + encoder, err := newEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { log.Error("failed to create row event encoder", zap.Error(err)) return nil, errors.Trace(err) @@ -214,6 +238,9 @@ func (g *encoderGroup) close() { encoder.Clean() } } + if g.claimCheck != nil { + g.claimCheck.Close() + } common.CleanMetrics(g.changefeedID) } diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index 0b6aa9d12e..10cb025b41 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -45,28 +45,41 @@ type batchEncoder struct { callbackBuff []func() claimCheck *claimcheck.ClaimCheck + cleanup func() config *common.Config } // NewBatchEncoder creates a new batchEncoder. -func NewBatchEncoder(ctx context.Context, config *common.Config) (common.EventEncoder, error) { - claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) - if err != nil { - return nil, errors.Trace(err) +func NewBatchEncoder( + ctx context.Context, config *common.Config, claimChecks ...*claimcheck.ClaimCheck, +) (common.EventEncoder, error) { + var claimCheck *claimcheck.ClaimCheck + if len(claimChecks) == 0 { + var err error + claimCheck, err = claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) + if err != nil { + return nil, errors.Trace(err) + } + } else { + claimCheck = claimChecks[0] } lock.Lock() clear(columnFlagsCache) lock.Unlock() - return &batchEncoder{ + encoder := &batchEncoder{ config: config, claimCheck: claimCheck, - }, nil + } + if len(claimChecks) == 0 && claimCheck != nil { + encoder.cleanup = claimCheck.Close + } + return encoder, nil } func (d *batchEncoder) Clean() { - if d.claimCheck != nil { - d.claimCheck.Close() + if d.cleanup != nil { + d.cleanup() } } diff --git a/pkg/sink/codec/simple/encoder.go b/pkg/sink/codec/simple/encoder.go index 9479478ae0..32292ec17b 100644 --- a/pkg/sink/codec/simple/encoder.go +++ b/pkg/sink/codec/simple/encoder.go @@ -28,24 +28,40 @@ type Encoder struct { messages []*common.Message config *common.Config claimCheck *claimcheck.ClaimCheck + cleanup func() marshaller marshaller } -func NewEncoder(ctx context.Context, config *common.Config) (common.EventEncoder, error) { - claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) - if err != nil { - return nil, errors.Trace(err) +func NewEncoder( + ctx context.Context, config *common.Config, claimChecks ...*claimcheck.ClaimCheck, +) (common.EventEncoder, error) { + var claimCheck *claimcheck.ClaimCheck + if len(claimChecks) == 0 { + var err error + claimCheck, err = claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID) + if err != nil { + return nil, errors.Trace(err) + } + } else { + claimCheck = claimChecks[0] } marshaller, err := newMarshaller(config) if err != nil { + if len(claimChecks) == 0 && claimCheck != nil { + claimCheck.Close() + } return nil, errors.Trace(err) } - return &Encoder{ + encoder := &Encoder{ messages: make([]*common.Message, 0, 1), config: config, claimCheck: claimCheck, marshaller: marshaller, - }, nil + } + if len(claimChecks) == 0 && claimCheck != nil { + encoder.cleanup = claimCheck.Close + } + return encoder, nil } // AppendRowChangedEvent implement the RowEventEncoder interface @@ -164,7 +180,7 @@ func (e *Encoder) EncodeDDLEvent(event *commonEvent.DDLEvent) (*common.Message, // CleanMetrics implement the RowEventEncoderBuilder interface func (e *Encoder) Clean() { - if e.claimCheck != nil { - e.claimCheck.Close() + if e.cleanup != nil { + e.cleanup() } } diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index fea8ba6cc9..870c6b914f 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -81,7 +81,8 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee }, nil } -// WriteMessage write message to the claim check external storage. +// WriteMessage writes a message to the claim-check external storage. +// It may be called concurrently. func (c *ClaimCheck) WriteMessage(ctx context.Context, key, value []byte, fileName string) (err error) { if !c.rawValue { m := common.ClaimCheckMessage{ diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go index e1f14ca877..b3e0bee671 100644 --- a/pkg/sink/kafka/claimcheck/claim_check_test.go +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -15,13 +15,16 @@ package claimcheck import ( "context" + "fmt" "testing" commonType "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/tidb/pkg/objstore" "github.com/pingcap/tidb/pkg/objstore/mockobjstore" "github.com/stretchr/testify/require" "go.uber.org/mock/gomock" + "golang.org/x/sync/errgroup" ) func TestClaimCheck(t *testing.T) { @@ -56,3 +59,34 @@ func TestClaimCheckCloseClosesStorage(t *testing.T) { claimCheck.Close() } + +func TestClaimCheckConcurrentWrites(t *testing.T) { + ctx := context.Background() + storage := objstore.NewMemStorage() + changefeedID := commonType.NewChangeFeedIDWithName("test", "default") + claimCheck := &ClaimCheck{ + storage: storage, + rawValue: true, + changefeedID: changefeedID, + metricSendMessageDuration: claimCheckSendMessageDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + metricSendMessageCount: claimCheckSendMessageCount.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + } + t.Cleanup(claimCheck.Close) + + const concurrency = 32 + group := new(errgroup.Group) + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + group.Go(func() error { + return claimCheck.WriteMessage(ctx, nil, []byte(fileName), fileName) + }) + } + require.NoError(t, group.Wait()) + + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + data, err := storage.ReadFile(ctx, fileName) + require.NoError(t, err) + require.Equal(t, fileName, string(data)) + } +} From 4cf73976f40c43252d3a866865e503638c4baab4 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Sat, 25 Jul 2026 00:07:27 +0800 Subject: [PATCH 3/9] simplify the code --- downstreamadapter/sink/kafka/sink.go | 61 ++++++++++------------- downstreamadapter/sink/kafka/sink_test.go | 44 ++++++---------- 2 files changed, 41 insertions(+), 64 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index c16642b66a..91abbf2e14 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -100,6 +100,18 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.Trace(err) } + claimCheck, err := claimcheck.New( + ctx, encoderConfig.LargeMessageHandle, encoderConfig.ChangefeedID, + ) + if err != nil { + return errors.Trace(err) + } + defer claimCheck.Close() + + if _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck); err != nil { + return errors.Trace(err) + } + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { return errors.Trace(err) @@ -114,57 +126,34 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.WrapError(errors.ErrKafkaNewProducer, err) } - encoderConfig.WithMaxMessageBytes(options.MaxMessageBytes). - WithMaxBatchedBytes(options.MaxBatchedBytes) - adminClient, err := factory.AdminClient(ctx) if err != nil { return errors.WrapError(errors.ErrKafkaNewProducer, err) } defer adminClient.Close() - return verifyKafkaSink(ctx, adminClient, topic, options.DeriveTopicConfig(), encoderConfig) -} - -func verifyKafkaSink( - ctx context.Context, - adminClient kafka.ClusterAdminClient, - topic string, - topicConfig *kafka.AutoCreateTopicConfig, - encoderConfig *common.Config, -) error { topics, err := adminClient.GetTopicsMeta([]string{topic}, false) if err != nil { return errors.Trace(err) } - if _, exists := topics[topic]; !exists { - if !topicConfig.AutoCreate { - return errors.ErrKafkaInvalidConfig.GenWithStack( - "`auto-create-topic` is false, and %s not found", topic) - } - - // The topic is not created, only validate its configuration and permissions. - err = adminClient.CreateTopic(&kafka.TopicDetail{ - Name: topic, - NumPartitions: topicConfig.PartitionNum, - ReplicationFactor: topicConfig.ReplicationFactor, - }, true) - if err != nil { - return errors.WrapError(errors.ErrKafkaCreateTopic, err) - } + if _, exists := topics[topic]; exists { + return nil } - claimCheck, err := claimcheck.New( - ctx, encoderConfig.LargeMessageHandle, encoderConfig.ChangefeedID, - ) - if err != nil { - return errors.Trace(err) + topicConfig := options.DeriveTopicConfig() + if !topicConfig.AutoCreate { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "`auto-create-topic` is false, and %s not found", topic) } - defer claimCheck.Close() - _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + // The topic is not created, only validate its configuration and permissions. + err = adminClient.CreateTopic(&kafka.TopicDetail{ + Name: topic, + NumPartitions: topicConfig.PartitionNum, + ReplicationFactor: topicConfig.ReplicationFactor, + }, true) if err != nil { - return errors.Trace(err) + return errors.WrapError(errors.ErrKafkaCreateTopic, err) } return nil } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 87085ccc0e..8e93386f34 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -37,7 +37,7 @@ import ( const kafkaSinkTestTopic = "mock_topic" -func TestVerifyValidatesEncoderConfigBeforeKafkaConnection(t *testing.T) { +func TestVerifyInvalidEncoderConfig(t *testing.T) { openProtocol := config.ProtocolOpen.String() sinkConfig := &config.SinkConfig{Protocol: &openProtocol} sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic + "?max-batch-size=0") @@ -50,6 +50,21 @@ func TestVerifyValidatesEncoderConfigBeforeKafkaConnection(t *testing.T) { require.ErrorContains(t, err, "invalid max-batch-size 0") } +func TestVerifyEncoderInitialization(t *testing.T) { + avroProtocol := config.ProtocolAvro.String() + schemaRegistry := "http://127.0.0.1:1" + sinkConfig := &config.SinkConfig{ + Protocol: &avroProtocol, + SchemaRegistry: &schemaRegistry, + } + sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic) + require.NoError(t, err) + + changefeedID := common.NewChangefeedID4Test("test", "verify-encoder") + err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig) + require.ErrorContains(t, err, "ErrAvroSchemaAPIError") +} + func newKafkaSinkForTestWithProducers(ctx context.Context, t *testing.T, ctrl *gomock.Controller, @@ -260,30 +275,3 @@ func TestKafkaSinkBatchConfig(t *testing.T) { require.Equal(t, 4096, sink.BatchCount()) require.Zero(t, sink.BatchBytes()) } - -func TestVerifyKafkaSinkRejectsInvalidClaimCheckStorageForExistingTopic(t *testing.T) { - ctrl := gomock.NewController(t) - adminClient := kafka.NewMockClusterAdminClient(ctrl) - adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, false).Return( - map[string]kafka.TopicDetail{ - kafkaSinkTestTopic: { - Name: kafkaSinkTestTopic, - NumPartitions: 1, - ReplicationFactor: 1, - }, - }, nil) - - encoderConfig := codeccommon.NewConfig(config.ProtocolOpen) - encoderConfig.ChangefeedID = common.NewChangefeedID4Test("test", "test") - encoderConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck - encoderConfig.LargeMessageHandle.ClaimCheckStorageURI = "unsupported:///claim-check" - - err := verifyKafkaSink( - context.Background(), - adminClient, - kafkaSinkTestTopic, - &kafka.AutoCreateTopicConfig{AutoCreate: false}, - encoderConfig, - ) - require.Error(t, err) -} From 403c637db989a9e2baa8726fafa47dceed39e615 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Sat, 25 Jul 2026 00:27:28 +0800 Subject: [PATCH 4/9] simplify the code --- downstreamadapter/sink/kafka/sink.go | 19 ++++++------------- pkg/sink/kafka/admin_test.go | 2 +- pkg/sink/kafka/options.go | 1 + 3 files changed, 8 insertions(+), 14 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 91abbf2e14..dae2b4ebd8 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -89,22 +89,16 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. options.Topic = topic encoderConfig, err := helper.GetEncoderConfig( - commonType.NewChangefeedID(changefeedID.Keyspace()), - uri, - protocol, - sinkConfig, - options.MaxMessageBytes, - options.MaxBatchedBytes, + changefeedID, uri, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, ) if err != nil { return errors.Trace(err) } - claimCheck, err := claimcheck.New( - ctx, encoderConfig.LargeMessageHandle, encoderConfig.ChangefeedID, - ) + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) if err != nil { - return errors.Trace(err) + return err } defer claimCheck.Close() @@ -142,11 +136,10 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. topicConfig := options.DeriveTopicConfig() if !topicConfig.AutoCreate { - return errors.ErrKafkaInvalidConfig.GenWithStack( - "`auto-create-topic` is false, and %s not found", topic) + return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) } - // The topic is not created, only validate its configuration and permissions. + // the topic is not created, only validate. err = adminClient.CreateTopic(&kafka.TopicDetail{ Name: topic, NumPartitions: topicConfig.PartitionNum, diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 67e20e565c..493e0c1bc3 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -64,7 +64,7 @@ func TestAdminClientClose(t *testing.T) { } } -func TestGetTopicsMetaUsesSmallestReplicationFactor(t *testing.T) { +func TestTopicReplicationFactor(t *testing.T) { ctrl := gomock.NewController(t) admin := NewMocksaramaClusterAdmin(ctrl) admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index faada6c8f9..2eb7ec550a 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -588,6 +588,7 @@ func adjustOptions( if err != nil { return errors.Trace(err) } + if err = validateRequiredAcks(ctx, admin, topics, topic, options); err != nil { return errors.Trace(err) } From 380b5e925579c531c1230827dd59eb59b138f041 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Mon, 27 Jul 2026 11:26:29 +0800 Subject: [PATCH 5/9] adjust when the topic not found --- downstreamadapter/sink/kafka/sink.go | 34 +++++++++--------- downstreamadapter/sink/kafka/sink_test.go | 44 ++++++++++++++--------- 2 files changed, 44 insertions(+), 34 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index dae2b4ebd8..9a544d69fb 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -102,10 +102,6 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. } defer claimCheck.Close() - if _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck); err != nil { - return errors.Trace(err) - } - isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { return errors.Trace(err) @@ -130,23 +126,25 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. if err != nil { return errors.Trace(err) } - if _, exists := topics[topic]; exists { - return nil - } + if _, exists := topics[topic]; !exists { + topicConfig := options.DeriveTopicConfig() + if !topicConfig.AutoCreate { + return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) + } - topicConfig := options.DeriveTopicConfig() - if !topicConfig.AutoCreate { - return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) + // the topic is not created, only validate. + err = adminClient.CreateTopic(&kafka.TopicDetail{ + Name: topic, + NumPartitions: topicConfig.PartitionNum, + ReplicationFactor: topicConfig.ReplicationFactor, + }, true) + if err != nil { + return errors.WrapError(errors.ErrKafkaCreateTopic, err) + } } - // the topic is not created, only validate. - err = adminClient.CreateTopic(&kafka.TopicDetail{ - Name: topic, - NumPartitions: topicConfig.PartitionNum, - ReplicationFactor: topicConfig.ReplicationFactor, - }, true) - if err != nil { - return errors.WrapError(errors.ErrKafkaCreateTopic, err) + if _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck); err != nil { + return errors.Trace(err) } return nil } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 8e93386f34..892b3e7c92 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -16,10 +16,13 @@ package kafka import ( "context" "fmt" + "net/http" + "net/http/httptest" "net/url" "testing" "time" + "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" @@ -37,30 +40,39 @@ import ( const kafkaSinkTestTopic = "mock_topic" -func TestVerifyInvalidEncoderConfig(t *testing.T) { - openProtocol := config.ProtocolOpen.String() - sinkConfig := &config.SinkConfig{Protocol: &openProtocol} - sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic + "?max-batch-size=0") - require.NoError(t, err) +func TestVerifyInvalidConfig(t *testing.T) { + broker := sarama.NewMockBroker(t, 1) + defer broker.Close() + broker.SetHandlerByMap(map[string]sarama.MockResponse{ + "ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys( + []sarama.ApiVersionsResponseKey{ + {ApiKey: 0}, + {ApiKey: 1}, + {ApiKey: 2}, + {ApiKey: 3, MaxVersion: 9}, + }), + "MetadataRequest": sarama.NewMockMetadataResponse(t). + SetController(broker.BrokerID()). + SetBroker(broker.Addr(), broker.BrokerID()). + SetLeader(kafkaSinkTestTopic, 0, broker.BrokerID()), + "DescribeConfigsRequest": sarama.NewMockDescribeConfigsResponse(t), + }) - changefeedID := common.NewChangefeedID4Test("test", "verify-existing-topic") - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - err = Verify(ctx, changefeedID, sinkURI, sinkConfig) - require.ErrorContains(t, err, "invalid max-batch-size 0") -} + schemaRegistry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "invalid response", http.StatusInternalServerError) + })) + defer schemaRegistry.Close() -func TestVerifyEncoderInitialization(t *testing.T) { avroProtocol := config.ProtocolAvro.String() - schemaRegistry := "http://127.0.0.1:1" sinkConfig := &config.SinkConfig{ Protocol: &avroProtocol, - SchemaRegistry: &schemaRegistry, + SchemaRegistry: &schemaRegistry.URL, } - sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic) + sinkURI, err := url.Parse("kafka://" + broker.Addr() + "/" + kafkaSinkTestTopic + + "?required-acks=1&kafka-version=2.4.0") require.NoError(t, err) - changefeedID := common.NewChangefeedID4Test("test", "verify-encoder") + changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config") err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig) require.ErrorContains(t, err, "ErrAvroSchemaAPIError") } From 3b77998340912e4a2601e5278c29257594ba6b30 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Mon, 27 Jul 2026 11:41:24 +0800 Subject: [PATCH 6/9] adjust code --- downstreamadapter/sink/kafka/sink.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 9a544d69fb..5a7c1dba22 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -143,7 +143,8 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. } } - if _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck); err != nil { + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + if err != nil { return errors.Trace(err) } return nil From 446e9eca4f1413b8a94323a1dc662a0a4defd9ba Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Mon, 27 Jul 2026 12:01:58 +0800 Subject: [PATCH 7/9] do not verify replication-factor when the topic exist --- pkg/sink/kafka/admin.go | 18 ++-------- pkg/sink/kafka/admin_test.go | 24 ------------- pkg/sink/kafka/options.go | 28 ++++------------ pkg/sink/kafka/options_test.go | 61 +++++++++++----------------------- 4 files changed, 28 insertions(+), 103 deletions(-) diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go index 0c58d8933f..16087c889c 100644 --- a/pkg/sink/kafka/admin.go +++ b/pkg/sink/kafka/admin.go @@ -146,27 +146,13 @@ func (a *saramaAdminClient) GetTopicsMeta(topics []string, ignoreTopicError bool continue } result[meta.Name] = TopicDetail{ - Name: meta.Name, - NumPartitions: int32(len(meta.Partitions)), - ReplicationFactor: minReplicationFactor(meta.Partitions), + Name: meta.Name, + NumPartitions: int32(len(meta.Partitions)), } } return result, nil } -func minReplicationFactor(partitions []*sarama.PartitionMetadata) int16 { - minReplicas := 0 - for i, partition := range partitions { - if partition == nil { - return 0 - } - if i == 0 || len(partition.Replicas) < minReplicas { - minReplicas = len(partition.Replicas) - } - } - return int16(minReplicas) -} - // IsAdminAuthorizationFailed checks whether err is an authorization failure from Kafka admin APIs. func IsAdminAuthorizationFailed(err error) bool { return errors.Is(err, sarama.ErrTopicAuthorizationFailed) || diff --git a/pkg/sink/kafka/admin_test.go b/pkg/sink/kafka/admin_test.go index 493e0c1bc3..c2e3f90e37 100644 --- a/pkg/sink/kafka/admin_test.go +++ b/pkg/sink/kafka/admin_test.go @@ -16,7 +16,6 @@ package kafka import ( "testing" - "github.com/IBM/sarama" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/pkg/common" "github.com/stretchr/testify/require" @@ -63,26 +62,3 @@ func TestAdminClientClose(t *testing.T) { }) } } - -func TestTopicReplicationFactor(t *testing.T) { - ctrl := gomock.NewController(t) - admin := NewMocksaramaClusterAdmin(ctrl) - admin.EXPECT().DescribeTopics([]string{"test-topic"}).Return([]*sarama.TopicMetadata{ - { - Name: "test-topic", - Err: sarama.ErrNoError, - Partitions: []*sarama.PartitionMetadata{ - {Replicas: []int32{1, 2, 3}}, - {Replicas: []int32{1, 2}}, - }, - }, - }, nil) - - client := &saramaAdminClient{ - changefeed: common.NewChangeFeedIDWithName("test", "default"), - admin: admin, - } - topics, err := client.GetTopicsMeta([]string{"test-topic"}, false) - require.NoError(t, err) - require.Equal(t, int16(2), topics["test-topic"].ReplicationFactor) -} diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 2eb7ec550a..b89e993fc9 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -589,8 +589,12 @@ func adjustOptions( return errors.Trace(err) } - if err = validateRequiredAcks(ctx, admin, topics, topic, options); err != nil { - return errors.Trace(err) + if _, exists := topics[topic]; !exists { + if err = validateMinInsyncReplicas( + ctx, admin, topics, topic, int(options.ReplicationFactor), + ); err != nil { + return errors.Trace(err) + } } return adjustTopicOptions(ctx, changefeedID, admin, options, topic, topics) } @@ -620,26 +624,6 @@ func adjustTopicOptions( return nil } -func validateRequiredAcks( - ctx context.Context, - admin ClusterAdminClient, - topics map[string]TopicDetail, - topic string, - options *options, -) error { - // Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1. - // If we don't check it, the producer probably can not send message to the topic. - // Because it will wait for the ack from all replicas. But we do not have enough replicas. - if options.RequiredAcks != WaitForAll { - return nil - } - replicationFactor := options.ReplicationFactor - if info, exists := topics[topic]; exists { - replicationFactor = info.ReplicationFactor - } - return validateMinInsyncReplicas(ctx, admin, topics, topic, int(replicationFactor)) -} - func adjustExistingTopicOption( ctx context.Context, changefeedID common.ChangeFeedID, diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index d33a9f21da..7e9f0673f0 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -83,11 +83,7 @@ func newKafkaAdminFixture(t *testing.T) *kafkaAdminFixture { } func (f *kafkaAdminFixture) addTopic(name string, partitionNum int32) { - f.topics[name] = TopicDetail{ - Name: name, - NumPartitions: partitionNum, - ReplicationFactor: mockClusterReplicationFactor, - } + f.topics[name] = TopicDetail{Name: name, NumPartitions: partitionNum} } func (f *kafkaAdminFixture) getTopicsMeta( @@ -454,9 +450,8 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t adminClient := adminFixture.admin detail := &TopicDetail{ - Name: topicName, - NumPartitions: 3, - ReplicationFactor: mockClusterReplicationFactor, + Name: topicName, + NumPartitions: 3, } err := adminClient.CreateTopic(detail, false) require.NoError(t, err) @@ -520,6 +515,21 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) { errors.Cause(err), ) + // required-acks does not affect validation for a topic to be created. + options.RequiredAcks = WaitForLocal + err = adjustOptions( + ctx, + changefeedID, + adminClient, + options, + "create-new-fail-with-local-acks", + ) + require.Regexp( + t, + ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker.*", + errors.Cause(err), + ) + // topic not exist, and `min.insync.replicas` not found in broker's configuration adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName) topicName := "no-topic-no-min-insync-replicas" @@ -545,42 +555,11 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) { err = adjustOptions(ctx, changefeedID, adminClient, options, topicName) require.Nil(t, err) - // Existing topics use their actual replication factor rather than the option - // used only when creating a topic. + // Existing topics are not validated against the replication factor used only + // when creating a topic. adminFixture.setMinInsyncReplicas("2") err = adjustOptions(ctx, changefeedID, adminClient, options, defaultMockTopicName) require.NoError(t, err) - - topicDetail := adminFixture.topics[defaultMockTopicName] - topicDetail.ReplicationFactor = 1 - adminFixture.topics[defaultMockTopicName] = topicDetail - err = adjustOptions(ctx, changefeedID, adminClient, options, defaultMockTopicName) - require.Regexp(t, - ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of topic.*", - errors.Cause(err), - ) -} - -func TestSkipAdjustConfigMinInsyncReplicasWhenRequiredAcksIsNotWailAll(t *testing.T) { - adminFixture := newKafkaAdminFixture(t) - adminClient := adminFixture.admin - - options := NewOptions() - options.BrokerEndpoints = []string{"127.0.0.1:9092"} - options.RequiredAcks = WaitForLocal - - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") - - // Do not report an error if the replication-factor is less than min.insync.replicas(1<2). - adminFixture.setMinInsyncReplicas("2") - err := adjustOptions( - context.Background(), - changefeedID, - adminClient, - options, - "skip-check-min-insync-replicas", - ) - require.Nil(t, err, "Should not report an error when `required-acks` is not `all`") } func TestCreateProducerFailed(t *testing.T) { From dc633ed534bd4d4ea8f953383881fd82f7ca7fde Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Mon, 27 Jul 2026 17:37:51 +0800 Subject: [PATCH 8/9] update code --- downstreamadapter/sink/kafka/sink.go | 3 + .../sink/topicmanager/kafka_topic_manager.go | 3 + .../topicmanager/kafka_topic_manager_test.go | 42 +++++- pkg/sink/kafka/options.go | 137 ++++++------------ pkg/sink/kafka/options_test.go | 93 +++++------- pkg/sink/kafka/sarama_factory.go | 2 +- 6 files changed, 123 insertions(+), 157 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 5a7c1dba22..0aee34e123 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -131,6 +131,9 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. if !topicConfig.AutoCreate { return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic) } + if err = topicConfig.ValidateReplicationFactor(adminClient); err != nil { + return err + } // the topic is not created, only validate. err = adminClient.CreateTopic(&kafka.TopicDetail{ diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index c80c7d65a1..4b9f3ba820 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -243,6 +243,9 @@ func (m *kafkaTopicManager) createTopic( fmt.Sprintf("`auto-create-topic` is false, "+ "and %s not found", topicName)) } + if err := m.cfg.ValidateReplicationFactor(m.admin); err != nil { + return 0, err + } start := time.Now() err := m.admin.CreateTopic(&kafka.TopicDetail{ diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go index 521b139c52..45ee322c63 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go @@ -82,6 +82,7 @@ func TestCreateTopic(t *testing.T) { AutoCreate: true, PartitionNum: 2, ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, } changefeedID := common.NewChangefeedID4Test("test", "test") @@ -137,6 +138,7 @@ func TestCreateTopic(t *testing.T) { require.NoError(t, err) require.Equal(t, int32(2), partitionNum) + cfg.RequiredAcks = kafka.WaitForLocal partitionNum, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic") require.NoError(t, err) require.Equal(t, int32(2), partitionNum) @@ -151,7 +153,12 @@ func TestCreateTopic(t *testing.T) { require.Equal(t, int32(2), partitionsNum) // Try to create a topic without auto create. - cfg.AutoCreate = false + cfg = &kafka.AutoCreateTopicConfig{ + AutoCreate: false, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + } manager = newKafkaTopicManager(ctx, "new-topic2", changefeedID, adminClient, cfg) defer manager.Close() _, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic2") @@ -182,6 +189,39 @@ func TestCreateTopic(t *testing.T) { require.False(t, gotFailedTopicValidateOnly) } +func TestCreateTopicValidatesReplicationFactor(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + adminClient := kafka.NewMockClusterAdminClient(ctrl) + topic := "new-topic" + 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().GetBrokerConfig(kafka.MinInsyncReplicasConfigName). + Return("2", nil), + ) + + manager := newKafkaTopicManager( + context.Background(), + topic, + common.NewChangefeedID4Test("test", "test"), + adminClient, + &kafka.AutoCreateTopicConfig{ + AutoCreate: true, + PartitionNum: 2, + ReplicationFactor: 1, + RequiredAcks: kafka.WaitForAll, + }, + ) + defer manager.Close() + + _, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), topic) + require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker") +} + func TestCreateTopicWaitsUntilVisible(t *testing.T) { t.Parallel() diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index b89e993fc9..27d68a020e 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -14,7 +14,6 @@ package kafka import ( - "context" "encoding/base64" "fmt" "net/http" @@ -244,6 +243,10 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, } if urlParameter.ReplicationFactor != nil { + if *urlParameter.ReplicationFactor <= 0 { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "invalid replication-factor %d", *urlParameter.ReplicationFactor) + } o.ReplicationFactor = *urlParameter.ReplicationFactor } @@ -536,11 +539,12 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf return nil } -// AutoCreateTopicConfig is used to create topic configuration. +// AutoCreateTopicConfig contains settings used to create and validate a topic. type AutoCreateTopicConfig struct { AutoCreate bool PartitionNum int32 ReplicationFactor int16 + RequiredAcks RequiredAcks } func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { @@ -548,9 +552,42 @@ func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig { AutoCreate: o.AutoCreate, PartitionNum: o.PartitionNum, ReplicationFactor: o.ReplicationFactor, + RequiredAcks: o.RequiredAcks, } } +// ValidateReplicationFactor checks whether a topic created with this config +// can satisfy the configured acknowledgment requirement. +func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClient) error { + if c.RequiredAcks != WaitForAll { + return nil + } + + raw, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) + if err != nil { + log.Warn("cannot get Kafka broker configuration, assume replication factor is valid", + zap.String("configName", MinInsyncReplicasConfigName), + zap.Int16("replicationFactor", c.ReplicationFactor), + zap.Error(err)) + return nil + } + minInsyncReplicas, err := strconv.Atoi(raw) + if err != nil { + return err + } + + if int(c.ReplicationFactor) < minInsyncReplicas { + return errors.ErrKafkaInvalidConfig.GenWithStack( + "TiCDC Kafka sink's `request.required.acks` defaults to -1, "+ + "TiCDC cannot deliver messages when the `replication-factor` %d "+ + "is smaller than the `min.insync.replicas` %d of broker", + c.ReplicationFactor, minInsyncReplicas, + ) + } + + return nil +} + var ( validClientID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`) commonInvalidChar = regexp.MustCompile(`[\?:,"]`) @@ -578,7 +615,6 @@ func NewKafkaClientID(captureAddr string, // It overwrites MaxMessageBytes with the final producer message limit derived // from the topic or broker configuration. func adjustOptions( - ctx context.Context, changefeedID common.ChangeFeedID, admin ClusterAdminClient, options *options, @@ -589,30 +625,11 @@ func adjustOptions( return errors.Trace(err) } - if _, exists := topics[topic]; !exists { - if err = validateMinInsyncReplicas( - ctx, admin, topics, topic, int(options.ReplicationFactor), - ); err != nil { - return errors.Trace(err) - } - } - return adjustTopicOptions(ctx, changefeedID, admin, options, topic, topics) -} - -func adjustTopicOptions( - ctx context.Context, - changefeedID common.ChangeFeedID, - admin ClusterAdminClient, - options *options, - topic string, - topics map[string]TopicDetail, -) error { info, exists := topics[topic] // once we have found the topic, no matter `auto-create-topic`, // make sure user input parameters are valid. - var err error if exists { - err = adjustExistingTopicOption(ctx, changefeedID, admin, options, topic, info) + err = adjustExistingTopicOption(changefeedID, admin, options, topic, info) } else { adjustNewTopicOptions(admin, changefeedID, options, topic) } @@ -625,14 +642,13 @@ func adjustTopicOptions( } func adjustExistingTopicOption( - ctx context.Context, changefeedID common.ChangeFeedID, admin ClusterAdminClient, options *options, topic string, info TopicDetail, ) error { - maxMessageBytes, err := getTopicMaxMessageBytes(ctx, admin, info.Name) + maxMessageBytes, err := getTopicMaxMessageBytes(admin, info.Name) if err != nil { log.Warn("`max.message.bytes` not found from topic's configuration, use the option `MaxMessageBytes` as default", zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()), @@ -682,12 +698,11 @@ func adjustNewTopicOptions( } func getTopicMaxMessageBytes( - ctx context.Context, admin ClusterAdminClient, topic string, ) (int, error) { raw, err := getTopicConfig( - ctx, admin, topic, + admin, topic, TopicMaxMessageBytesConfigName, BrokerMessageMaxBytesConfigName, ) @@ -713,79 +728,11 @@ func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, error) { return messageMaxBytes, nil } -func validateMinInsyncReplicas( - ctx context.Context, - admin ClusterAdminClient, - topics map[string]TopicDetail, - topic string, - replicationFactor int, -) error { - minInsyncReplicasConfigGetter := func() (string, bool, error) { - info, exists := topics[topic] - if exists { - minInsyncReplicasStr, err := getTopicConfig( - ctx, admin, info.Name, - MinInsyncReplicasConfigName, - MinInsyncReplicasConfigName) - if err != nil { - return "", true, err - } - return minInsyncReplicasStr, true, nil - } - - minInsyncReplicasStr, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName) - if err != nil { - return "", false, err - } - - return minInsyncReplicasStr, false, nil - } - - minInsyncReplicasStr, exists, err := minInsyncReplicasConfigGetter() - if err != nil { - // 'min.insync.replica' is invisible to us in Confluent Cloud Kafka. - if errors.ErrKafkaConfigNotFound.Equal(err) { - log.Warn("TiCDC cannot find `min.insync.replicas` from broker's configuration, " + - "please make sure that the replication factor is greater than or equal " + - "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.") - } - 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 { - return err - } - - configFrom := "topic" - if !exists { - configFrom = "broker" - } - - if replicationFactor < minInsyncReplicas { - msg := fmt.Sprintf("`replication-factor` cannot be smaller than the `%s` of %s", - MinInsyncReplicasConfigName, configFrom) - log.Error(msg, zap.Int("replication-factor", replicationFactor), - zap.Int("min.insync.replicas", minInsyncReplicas)) - return errors.ErrKafkaInvalidConfig.GenWithStack( - "TiCDC Kafka sink's `request.required.acks` defaults to -1, "+ - "TiCDC cannot deliver messages when the `replication-factor` %d "+ - "is smaller than the `min.insync.replicas` %d of %s", - replicationFactor, minInsyncReplicas, configFrom, - ) - } - - return nil -} - // getTopicConfig gets topic config by name. // If the topic does not have this configuration, // we will try to get it from the broker's configuration. // NOTICE: The configuration names of topic and broker may be different for the same configuration. func getTopicConfig( - ctx context.Context, admin ClusterAdminClient, topicName string, topicConfigName string, diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go index 7e9f0673f0..d922166c40 100644 --- a/pkg/sink/kafka/options_test.go +++ b/pkg/sink/kafka/options_test.go @@ -205,6 +205,18 @@ func TestCompleteOptions(t *testing.T) { options = NewOptions() err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Regexp(t, ".*invalid syntax.*", errors.Cause(err)) + for _, replicationFactor := range []string{"0", "-1"} { + uri = "kafka://127.0.0.1:9092/abc?replication-factor=" + replicationFactor + sinkURI, err = url.Parse(uri) + require.NoError(t, err) + options = NewOptions() + err = options.Apply( + commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), + sinkURI, + config.GetDefaultReplicaConfig().Sink, + ) + require.ErrorContains(t, err, "invalid replication-factor "+replicationFactor) + } // Illegal max-message-bytes. uri = "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&max-message-bytes=a" @@ -471,7 +483,7 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t expectedProducerLimit := adminFixture.brokerMessageMaxBytes() ctx := context.Background() - err = adjustOptions(ctx, changefeedID, adminClient, options, topicName) + err = adjustOptions(changefeedID, adminClient, options, topicName) require.NoError(t, err) saramaConfig, err := newSaramaConfig(ctx, options) @@ -489,76 +501,38 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t } } -func TestAdjustConfigMinInsyncReplicas(t *testing.T) { +func TestValidateReplicationFactor(t *testing.T) { adminFixture := newKafkaAdminFixture(t) adminClient := adminFixture.admin - - options := NewOptions() - options.BrokerEndpoints = []string{"127.0.0.1:9092"} - - // Report an error if the replication-factor is less than min.insync.replicas - // when the topic does not exist. adminFixture.setMinInsyncReplicas("2") - changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") - ctx := context.Background() - err := adjustOptions( - ctx, - changefeedID, - adminClient, - options, - "create-new-fail-invalid-min-insync-replicas", - ) + topicConfig := &AutoCreateTopicConfig{ + AutoCreate: true, + ReplicationFactor: 1, + RequiredAcks: WaitForAll, + } + err := topicConfig.ValidateReplicationFactor(adminClient) require.Regexp( t, ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker.*", errors.Cause(err), ) - // required-acks does not affect validation for a topic to be created. - options.RequiredAcks = WaitForLocal - err = adjustOptions( - ctx, - changefeedID, - adminClient, - options, - "create-new-fail-with-local-acks", - ) - require.Regexp( - t, - ".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker.*", - errors.Cause(err), - ) + localAcksConfig := &AutoCreateTopicConfig{ + AutoCreate: true, + ReplicationFactor: 1, + RequiredAcks: WaitForLocal, + } + err = localAcksConfig.ValidateReplicationFactor(adminClient) + require.NoError(t, err) - // topic not exist, and `min.insync.replicas` not found in broker's configuration adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName) - topicName := "no-topic-no-min-insync-replicas" - err = adjustOptions(ctx, changefeedID, adminClient, options, "no-topic-no-min-insync-replicas") - require.Nil(t, err) - err = adminClient.CreateTopic(&TopicDetail{ - Name: topicName, + missingBrokerConfig := &AutoCreateTopicConfig{ + AutoCreate: true, ReplicationFactor: 1, - }, false) - require.ErrorIs(t, err, sarama.ErrPolicyViolation) - - // Report an error if the replication-factor is less than min.insync.replicas - // when the topic does exist. - - // topic exist, but `min.insync.replicas` not found in topic and broker configuration - topicName = "topic-no-options-entry" - err = adminClient.CreateTopic(&TopicDetail{ - Name: topicName, - ReplicationFactor: 3, - NumPartitions: 3, - }, false) - require.Nil(t, err) - err = adjustOptions(ctx, changefeedID, adminClient, options, topicName) - require.Nil(t, err) - - // Existing topics are not validated against the replication factor used only - // when creating a topic. - adminFixture.setMinInsyncReplicas("2") - err = adjustOptions(ctx, changefeedID, adminClient, options, defaultMockTopicName) + RequiredAcks: WaitForAll, + } + err = missingBrokerConfig.ValidateReplicationFactor(adminClient) require.NoError(t, err) } @@ -729,7 +703,6 @@ func TestConfigurationCombinations(t *testing.T) { sinkURI, err := url.Parse(uri) require.Nil(t, err) - ctx := context.Background() options := NewOptions() err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink) require.Nil(t, err) @@ -745,7 +718,7 @@ func TestConfigurationCombinations(t *testing.T) { } changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test") - err = adjustOptions(ctx, changefeedID, adminClient, options, topic) + err = adjustOptions(changefeedID, adminClient, options, topic) require.Nil(t, err) require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes) require.Equal( diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go index 39c69c0c0b..3be5fd841a 100644 --- a/pkg/sink/kafka/sarama_factory.go +++ b/pkg/sink/kafka/sarama_factory.go @@ -57,7 +57,7 @@ func NewSaramaFactory( admin.Close() }() - if err = adjustOptions(ctx, changefeedID, admin, o, o.Topic); err != nil { + if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil { return nil, errors.Trace(err) } From 3c541a8fe34b5b182fe3f72e5f93dc667b16325d Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Mon, 27 Jul 2026 17:44:39 +0800 Subject: [PATCH 9/9] update code --- downstreamadapter/sink/topicmanager/kafka_topic_manager.go | 6 ++---- pkg/sink/kafka/options.go | 3 +-- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go index 4b9f3ba820..6d70c84b52 100644 --- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go +++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go @@ -15,7 +15,6 @@ package topicmanager import ( "context" - "fmt" "sync" "time" @@ -239,10 +238,9 @@ func (m *kafkaTopicManager) createTopic( topicName string, ) (int32, error) { if !m.cfg.AutoCreate { - return 0, errors.ErrKafkaInvalidConfig.GenWithStack( - fmt.Sprintf("`auto-create-topic` is false, "+ - "and %s not found", topicName)) + return 0, errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topicName) } + if err := m.cfg.ValidateReplicationFactor(m.admin); err != nil { return 0, err } diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go index 27d68a020e..506e6287ef 100644 --- a/pkg/sink/kafka/options.go +++ b/pkg/sink/kafka/options.go @@ -244,8 +244,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID, if urlParameter.ReplicationFactor != nil { if *urlParameter.ReplicationFactor <= 0 { - return errors.ErrKafkaInvalidConfig.GenWithStack( - "invalid replication-factor %d", *urlParameter.ReplicationFactor) + return errors.ErrKafkaInvalidConfig.GenWithStack("invalid replication-factor %d", *urlParameter.ReplicationFactor) } o.ReplicationFactor = *urlParameter.ReplicationFactor }