From 980e28cdf45ac8b138e6057f7ee92404c5b1d7cb Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 23 Jul 2026 19:29:36 +0800 Subject: [PATCH 1/4] 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/4] 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/4] 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/4] 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) }