From 45dfe1607fe4b8ffc65648d5731de136edd86a70 Mon Sep 17 00:00:00 2001 From: Ling Jin <7138436+3AceShowHand@users.noreply.github.com> Date: Fri, 24 Jul 2026 23:17:26 +0800 Subject: [PATCH] This is an automated cherry-pick of #5718 Signed-off-by: ti-chi-bot --- downstreamadapter/sink/kafka/helper.go | 73 +++++++----- downstreamadapter/sink/kafka/sink.go | 104 +++++++++++++++-- downstreamadapter/sink/kafka/sink_test.go | 75 ++++++++++++ downstreamadapter/sink/pulsar/helper.go | 4 +- pkg/sink/codec/avro/arvo.go | 2 - pkg/sink/codec/bootstraper.go | 1 - pkg/sink/codec/builder.go | 9 +- pkg/sink/codec/canal/canal_json_encoder.go | 12 +- .../codec/canal/canal_json_encoder_test.go | 50 ++++---- pkg/sink/codec/canal/canal_json_test.go | 44 +++---- pkg/sink/codec/common/encoder.go | 2 - pkg/sink/codec/debezium/encoder.go | 2 - pkg/sink/codec/encoder_group.go | 11 +- pkg/sink/codec/open/encoder.go | 12 +- pkg/sink/codec/open/encoder_test.go | 108 +++++++++++++----- pkg/sink/codec/simple/encoder.go | 13 +-- pkg/sink/codec/simple/encoder_test.go | 53 ++++++--- pkg/sink/kafka/claimcheck/claim_check.go | 22 ++-- pkg/sink/kafka/claimcheck/claim_check_test.go | 96 ++++++++++++++++ 19 files changed, 500 insertions(+), 193 deletions(-) create mode 100644 pkg/sink/kafka/claimcheck/claim_check_test.go diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index c45b7eb10f..df2d61cf1d 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -27,6 +27,7 @@ import ( "github.com/pingcap/ticdc/pkg/sink/codec" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" + "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/tidb/br/pkg/utils" ) @@ -38,6 +39,7 @@ type components struct { topicManager topicmanager.TopicManager adminClient kafka.ClusterAdminClient factory kafka.Factory + claimCheck *claimcheck.ClaimCheck } func (c components) close() { @@ -47,6 +49,9 @@ func (c components) close() { if c.topicManager != nil { c.topicManager.Close() } + if c.claimCheck != nil { + c.claimCheck.Close() + } } func newKafkaSinkComponentWithFactory(ctx context.Context, @@ -55,78 +60,94 @@ func newKafkaSinkComponentWithFactory(ctx context.Context, sinkConfig *config.SinkConfig, factoryCreator kafka.FactoryCreator, ) (components, config.Protocol, error) { - kafkaComponent := components{} + var ( + comp components + err error + ) + // must release resources when error occurs. + defer func() { + if err != nil { + comp.close() + } + }() protocol, err := helper.GetProtocol(utils.GetOrZero(sinkConfig.Protocol)) if err != nil { - return kafkaComponent, config.ProtocolUnknown, errors.Trace(err) + return comp, config.ProtocolUnknown, errors.Trace(err) } topic, err := helper.GetTopic(sinkURI) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } options := kafka.NewOptions() if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil { - return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaInvalidConfig, err) + return comp, protocol, errors.WrapError(errors.ErrKafkaInvalidConfig, err) } options.Topic = topic +<<<<<<< HEAD kafkaComponent.factory, err = factoryCreator(ctx, options, changefeedID) +======= + comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) if err != nil { - return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) } +<<<<<<< HEAD kafkaComponent.eventRouter, err = eventrouter.NewEventRouter( sinkConfig, topic, false, protocol == config.ProtocolAvro) +======= + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro + comp.eventRouter, err = eventrouter.NewEventRouter( + sinkConfig, topic, false, isAvroLike) +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - kafkaComponent.columnSelector, err = columnselector.New(sinkConfig) + comp.columnSelector, err = columnselector.New(sinkConfig) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) + comp.claimCheck, err = claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig) + comp.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, comp.claimCheck, changefeedID) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - kafkaComponent.adminClient, err = kafkaComponent.factory.AdminClient(ctx) + comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck) if err != nil { - return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, errors.Trace(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() - } - }() + comp.adminClient, err = comp.factory.AdminClient(ctx) + if err != nil { + return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + } - kafkaComponent.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( + comp.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( ctx, changefeedID, topic, options.DeriveTopicConfig(), - kafkaComponent.adminClient, + comp.adminClient, ) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - return kafkaComponent, protocol, nil + return comp, protocol, nil } func newKafkaSinkComponent( diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index e6d1038b26..c68bba0f30 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -27,6 +27,7 @@ import ( "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/sink/kafka" + "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/ticdc/pkg/util" "github.com/pingcap/ticdc/utils/chann" "go.uber.org/atomic" @@ -68,9 +69,90 @@ func (s *sink) SinkType() commonType.SinkType { } func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error { +<<<<<<< HEAD comp, _, err := newKafkaSinkComponent(ctx, changefeedID, uri, sinkConfig) defer comp.close() return err +======= + protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol)) + if err != nil { + return errors.Trace(err) + } + + topic, err := helper.GetTopic(uri) + if err != nil { + return errors.Trace(err) + } + + options := kafka.NewOptions() + if err = options.Apply(changefeedID, uri, sinkConfig); err != nil { + return errors.WrapError(errors.ErrKafkaInvalidConfig, err) + } + options.Topic = topic + + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, uri, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return errors.Trace(err) + } + + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return err + } + defer claimCheck.Close() + + isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro + if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil { + return errors.Trace(err) + } + + if _, err = columnselector.New(sinkConfig); err != nil { + return errors.Trace(err) + } + + factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID) + if err != nil { + return errors.WrapError(errors.ErrKafkaNewProducer, err) + } + + adminClient, err := factory.AdminClient(ctx) + if err != nil { + return errors.WrapError(errors.ErrKafkaNewProducer, err) + } + defer adminClient.Close() + + 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) + } + + // 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) + } + + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + if err != nil { + return errors.Trace(err) + } + return nil +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) } func New( @@ -89,24 +171,30 @@ func newWithComponents( protocol config.Protocol, comp components, ) (*sink, error) { + statistics := metrics.NewStatistics(changefeedID, keyspaceID, "sink") var ( err error asyncProducer kafka.AsyncProducer syncProducer kafka.SyncProducer ) defer func() { - if err != nil { - if syncProducer != nil { - syncProducer.Close() - } - if asyncProducer != nil { - asyncProducer.Close() - } - comp.close() + if err == nil { + return + } + if syncProducer != nil { + syncProducer.Close() + } + if asyncProducer != nil { + asyncProducer.Close() } + comp.close() + statistics.Close() }() +<<<<<<< HEAD statistics := metrics.NewStatistics(changefeedID, "sink") +======= +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) asyncProducer, err = comp.factory.AsyncProducer(ctx) if err != nil { return nil, err diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 0bb4708f58..5799845d2d 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -51,7 +51,82 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, statistics := metrics.NewStatistics(changefeedID, "sink") comp, protocol, err := newKafkaSinkComponentForTest(ctx, changefeedID, sinkURI, sinkConfig) if err != nil { +<<<<<<< HEAD return nil, errors.Trace(err) +======= + return nil, err + } + topic, err := helper.GetTopic(sinkURI) + if err != nil { + return nil, err + } + options := kafka.NewOptions() + if err = options.Apply(changefeedID, sinkURI, sinkConfig); err != nil { + return nil, err + } + options.Topic = topic + + adminClient := kafka.NewMockClusterAdminClient(ctrl) + adminClient.EXPECT().GetTopicsMeta([]string{kafkaSinkTestTopic}, true).Return( + map[string]kafka.TopicDetail{ + kafkaSinkTestTopic: { + Name: kafkaSinkTestTopic, + NumPartitions: 1, + }, + }, nil) + adminClient.EXPECT().Close().AnyTimes() + + metricsCollector := kafka.NewMockMetricsCollector(ctrl) + metricsCollector.EXPECT().Run(gomock.Any()).AnyTimes() + + factory := kafka.NewMockFactory(ctrl) + factory.EXPECT().AsyncProducer(gomock.Any()).Return(asyncProducer, nil) + factory.EXPECT().SyncProducer(gomock.Any()).Return(syncProducer, nil) + factory.EXPECT().MetricsCollector(adminClient).Return(metricsCollector) + + eventRouter, err := eventrouter.NewEventRouter(sinkConfig, topic, false, false) + if err != nil { + return nil, err + } + columnSelector, err := columnselector.New(sinkConfig) + if err != nil { + return nil, err + } + encoderConfig, err := helper.GetEncoderConfig( + changefeedID, sinkURI, protocol, sinkConfig, + options.MaxMessageBytes, options.MaxBatchedBytes, + ) + if err != nil { + return nil, err + } + encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) + if err != nil { + return nil, err + } + encoder, err := codec.NewEventEncoder(ctx, encoderConfig, nil) + if err != nil { + return nil, err + } + topicManager, err := topicmanager.GetTopicManagerAndTryCreateTopic( + ctx, + changefeedID, + topic, + options.DeriveTopicConfig(), + adminClient, + ) + if err != nil { + return nil, err + } + + comp := components{ + encoderGroup: encoderGroup, + encoder: encoder, + columnSelector: columnSelector, + eventRouter: eventRouter, + topicManager: topicManager, + adminClient: adminClient, + factory: factory, +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) } // We must close adminClient when this func return cause by an error diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index acc1cddd38..fb7e5ca642 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -127,12 +127,12 @@ func newPulsarSinkComponentWithFactory(ctx context.Context, return pulsarComponent, protocol, errors.Trace(err) } - pulsarComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) + pulsarComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) if err != nil { return pulsarComponent, protocol, errors.Trace(err) } - pulsarComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig) + pulsarComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, nil) if err != nil { return pulsarComponent, protocol, errors.Trace(err) } diff --git a/pkg/sink/codec/avro/arvo.go b/pkg/sink/codec/avro/arvo.go index 515de064d9..cd15233bc8 100644 --- a/pkg/sink/codec/avro/arvo.go +++ b/pkg/sink/codec/avro/arvo.go @@ -698,8 +698,6 @@ func (a *BatchEncoder) columnToAvroData( } } -func (a *BatchEncoder) Clean() {} - type avroEncodeResult struct { data []byte // header is the message header, it will be encoder into the head diff --git a/pkg/sink/codec/bootstraper.go b/pkg/sink/codec/bootstraper.go index a57da8360f..9bff4368a8 100644 --- a/pkg/sink/codec/bootstraper.go +++ b/pkg/sink/codec/bootstraper.go @@ -79,7 +79,6 @@ func (b *bootstrapWorker) run(ctx context.Context) error { sendTicker := time.NewTicker(bootstrapWorkerTickerInterval) gcTicker := time.NewTicker(bootstrapWorkerGCInterval) defer func() { - b.rowEventEncoder.Clean() gcTicker.Stop() sendTicker.Stop() }() diff --git a/pkg/sink/codec/builder.go b/pkg/sink/codec/builder.go index 8a17146921..5bcb591232 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -28,21 +28,22 @@ 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) { +func NewEventEncoder(ctx context.Context, cfg *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { switch cfg.Protocol { case config.ProtocolDefault, config.ProtocolOpen: - return open.NewBatchEncoder(ctx, cfg) + return open.NewBatchEncoder(cfg, claimCheck) case config.ProtocolAvro: return avro.NewAvroEncoder(ctx, cfg) case config.ProtocolCanalJSON: - return canal.NewJSONRowEventEncoder(ctx, cfg) + return canal.NewJSONRowEventEncoder(cfg, claimCheck) case config.ProtocolDebezium: return debezium.NewBatchEncoder(cfg, config.GetGlobalServerConfig().ClusterID), nil case config.ProtocolSimple: - return simple.NewEncoder(ctx, cfg) + return simple.NewEncoder(cfg, claimCheck) 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 7425a666ef..dc076f3ee0 100644 --- a/pkg/sink/codec/canal/canal_json_encoder.go +++ b/pkg/sink/codec/canal/canal_json_encoder.go @@ -373,11 +373,7 @@ type JSONRowEventEncoder struct { } // 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(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { return &JSONRowEventEncoder{ messages: make([]*common.Message, 0, 1), config: config, @@ -582,9 +578,3 @@ func (c *JSONRowEventEncoder) EncodeDDLEvent(e *commonEvent.DDLEvent) (*common.M return common.NewMsg(nil, value), nil } - -func (c *JSONRowEventEncoder) Clean() { - if c.claimCheck != nil { - c.claimCheck.CleanMetrics() - } -} diff --git a/pkg/sink/codec/canal/canal_json_encoder_test.go b/pkg/sink/codec/canal/canal_json_encoder_test.go index 2953642f89..91e9f297f7 100644 --- a/pkg/sink/codec/canal/canal_json_encoder_test.go +++ b/pkg/sink/codec/canal/canal_json_encoder_test.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/stretchr/testify/require" ) @@ -47,7 +48,7 @@ func TestDMLE2E(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -131,7 +132,7 @@ func TestCanalJSONCompressionE2E(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4 ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -208,7 +209,7 @@ func TestEncodeRoutedDMLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -237,7 +238,7 @@ func TestEncodeRoutedDMLEventUsesTargetNames(t *testing.T) { func TestEncodeRoutedDDLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -269,8 +270,11 @@ func TestCanalJSONClaimCheckE2E(t *testing.T) { for _, rawValue := range []bool{false, true} { codecConfig.LargeMessageHandle.ClaimCheckRawValue = rawValue + claimCheck, err := claimcheck.New(ctx, codecConfig.LargeMessageHandle, codecConfig.ChangefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, claimCheck) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -315,9 +319,7 @@ func TestNewCanalJSONMessageHandleKeyOnly4LargeMessage(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4 codecConfig.MaxMessageBytes = 500 - ctx := context.Background() - - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -360,9 +362,8 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { defer helper.Close() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -382,7 +383,7 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { require.Equal(t, "CREATE", msg.EventType) codecConfig.EnableTiDBExtension = true - encIface, err = NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -397,9 +398,8 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { } func TestBatching(t *testing.T) { - ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) require.NotNil(t, encoder) @@ -434,13 +434,12 @@ func TestBatching(t *testing.T) { func TestEncodeCheckpointEvent(t *testing.T) { t.Parallel() - ctx := context.Background() var watermark uint64 = 2333 for _, enable := range []bool{false, true} { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = enable - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) msg, err := encoder.EncodeCheckpointEvent(watermark) @@ -482,9 +481,7 @@ func TestCheckpointEventValueMarshal(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - ctx := context.Background() - - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) var watermark uint64 = 1024 @@ -519,7 +516,7 @@ func TestDDLEventWithExtension(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) require.NotNil(t, encoder) @@ -561,9 +558,8 @@ func TestCanalJSONAppendRowChangedEventWithCallback(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - ctx := context.Background() - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) count := 0 @@ -654,7 +650,7 @@ func TestMaxMessageBytes(t *testing.T) { maxMessageBytes := 300 codecConfig := common.NewConfig(config.ProtocolCanalJSON).WithMaxMessageBytes(maxMessageBytes) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -669,7 +665,7 @@ func TestMaxMessageBytes(t *testing.T) { // the test message length is larger than max-message-bytes codecConfig = codecConfig.WithMaxMessageBytes(100) - encIface, err = NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -689,7 +685,7 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -737,7 +733,7 @@ func TestE2EPartitionTableByHash(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -794,7 +790,7 @@ func TestE2EPartitionTableByRange(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -858,7 +854,7 @@ func TestE2EPartitionTable(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) diff --git a/pkg/sink/codec/canal/canal_json_test.go b/pkg/sink/codec/canal/canal_json_test.go index 4fa008a27b..4c115bbfc9 100644 --- a/pkg/sink/codec/canal/canal_json_test.go +++ b/pkg/sink/codec/canal/canal_json_test.go @@ -24,6 +24,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/tidb/pkg/util/chunk" "github.com/stretchr/testify/require" ) @@ -67,7 +68,7 @@ func TestIntegerContentCompatible(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -151,7 +152,7 @@ func TestIntegerTypes(t *testing.T) { for _, enableTiDBExtension := range []bool{true, false} { for _, event := range []*commonEvent.RowEvent{minValueEvent, maxValueEvent} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) @@ -213,7 +214,7 @@ func TestFloatTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -262,7 +263,7 @@ func TestTimeTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -312,7 +313,7 @@ func TestStringTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -362,7 +363,7 @@ func TestBlobTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -412,7 +413,7 @@ func TestTextTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -471,7 +472,7 @@ func TestOtherTypes(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -529,7 +530,7 @@ func TestDMLEventWithColumnSelector(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -588,7 +589,7 @@ func TestDMLMultiplePK(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -674,7 +675,7 @@ func TestDMLMessageTooLarge(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig = codecConfig.WithMaxMessageBytes(300) codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(context.Background(), codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(context.Background(), "", rowEvent) require.ErrorIs(t, err, errors.ErrMessageTooLarge) @@ -772,7 +773,10 @@ func TestLargeMessageClaimCheck(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = "snappy" codecConfig.LargeMessageHandle.ClaimCheckStorageURI = "file:///tmp/canal-json-claim-check" - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + claimCheck, err := claimcheck.New(ctx, codecConfig.LargeMessageHandle, codecConfig.ChangefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) + encoder, err := NewJSONRowEventEncoder(codecConfig, claimCheck) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertEvent) @@ -863,7 +867,7 @@ func TestMessageLargeHandleKeyOnly(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionHandleKeyOnly codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -949,7 +953,7 @@ func TestDMLTypeEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -981,7 +985,7 @@ func TestDMLTypeEvent(t *testing.T) { // update with only updated columns codecConfig.OnlyOutputUpdatedColumns = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", updateEvent) @@ -1012,7 +1016,7 @@ func TestDDLSequence(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) @@ -1142,7 +1146,7 @@ func TestCreateTableDDL(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) message, err := encoder.EncodeDDLEvent(ddlEvent) @@ -1173,7 +1177,7 @@ func TestCreateTableDDL(t *testing.T) { func TestCheckpointTs(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) watermark := uint64(179394) @@ -1183,7 +1187,7 @@ func TestCheckpointTs(t *testing.T) { // with extension codecConfig.EnableTiDBExtension = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) message, err = encoder.EncodeCheckpointEvent(watermark) require.NoError(t, err) @@ -1243,7 +1247,7 @@ func TestRowKey(t *testing.T) { codecConfig.OnlyOutputUpdatedColumns = true codecConfig.EnableTiDBExtension = true codecConfig.OutputRowKey = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) require.NoError(t, err) diff --git a/pkg/sink/codec/common/encoder.go b/pkg/sink/codec/common/encoder.go index bcb9afd365..95bf191e59 100644 --- a/pkg/sink/codec/common/encoder.go +++ b/pkg/sink/codec/common/encoder.go @@ -31,8 +31,6 @@ type EventEncoder interface { AppendRowChangedEvent(context.Context, string, *commonEvent.RowEvent) error // Build builds the batch messages from AppendRowChangedEvent and returns the messages. Build() []*Message - // clean the resources - Clean() } // TxnEventEncoder is an abstraction for events encoder diff --git a/pkg/sink/codec/debezium/encoder.go b/pkg/sink/codec/debezium/encoder.go index c0f8c3d07a..8a78c6e9f4 100644 --- a/pkg/sink/codec/debezium/encoder.go +++ b/pkg/sink/codec/debezium/encoder.go @@ -165,8 +165,6 @@ func (d *BatchEncoder) Build() []*common.Message { return result } -func (d *BatchEncoder) Clean() {} - // newBatchEncoder creates a new Debezium BatchEncoder. func NewBatchEncoder(c *common.Config, clusterID string) common.EventEncoder { batch := &BatchEncoder{ diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go index 7ae503c985..eeeca21a20 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" @@ -67,18 +68,21 @@ func NewEncoderGroup( ctx context.Context, cfg *config.SinkConfig, encoderConfig *common.Config, + claimCheck *claimcheck.ClaimCheck, changefeedID commonType.ChangeFeedID, ) (*encoderGroup, error) { concurrency := util.GetOrZero(cfg.EncoderConcurrency) if concurrency <= 0 { concurrency = config.DefaultEncoderGroupConcurrency } + inputCh := make([]chan *future, concurrency) rowEventEncoders := make([]common.EventEncoder, concurrency) + var err error for i := 0; i < concurrency; i++ { inputCh[i] = make(chan *future, defaultInputChanSize) - rowEventEncoders[i], err = NewEventEncoder(ctx, encoderConfig) + 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) @@ -88,7 +92,7 @@ func NewEncoderGroup( var bw *bootstrapWorker 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) @@ -206,9 +210,6 @@ func (g *encoderGroup) Output() <-chan *future { func (g *encoderGroup) cleanMetrics() { encoderGroupInputChanSizeGauge.DeleteLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name()) - for _, encoder := range g.rowEventEncoders { - encoder.Clean() - } common.CleanMetrics(g.changefeedID) } diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index 75f82e9d03..567ce8b60f 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -50,11 +50,7 @@ type batchEncoder struct { } // 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(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { lock.Lock() clear(columnFlagsCache) lock.Unlock() @@ -64,12 +60,6 @@ func NewBatchEncoder(ctx context.Context, config *common.Config) (common.EventEn }, nil } -func (d *batchEncoder) Clean() { - if d.claimCheck != nil { - d.claimCheck.CleanMetrics() - } -} - func (d *batchEncoder) fetchColumnFlags(e *commonEvent.RowEvent) map[string]uint64 { lock.RLock() result, ok := columnFlagsCache[e.GetTableID()] diff --git a/pkg/sink/codec/open/encoder_test.go b/pkg/sink/codec/open/encoder_test.go index 9b02366709..45e48d7cbf 100644 --- a/pkg/sink/codec/open/encoder_test.go +++ b/pkg/sink/codec/open/encoder_test.go @@ -64,7 +64,7 @@ func TestEncodeFlag(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - enc, err := NewBatchEncoder(ctx, codecConfig) + enc, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = enc.AppendRowChangedEvent(ctx, "", insertEvent) @@ -153,7 +153,7 @@ func TestIntegerTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) for _, event := range []*commonEvent.RowEvent{minValueEvent, maxValueEvent} { - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) @@ -209,7 +209,7 @@ func TestFloatTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -258,7 +258,7 @@ func TestTimeTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -307,7 +307,7 @@ func TestStringTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -357,7 +357,7 @@ func TestBlobTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -407,7 +407,7 @@ func TestTextTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -454,7 +454,7 @@ func TestVectorType(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -503,7 +503,7 @@ func TestCollation(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -561,7 +561,7 @@ func TestOtherTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -588,7 +588,7 @@ func TestOtherTypes(t *testing.T) { func TestEncodeCheckpoint(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolOpen) ctx := context.Background() - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) checkpoint := uint64(12345678) @@ -629,7 +629,7 @@ func TestCreateTableDDL(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) message, err := encoder.EncodeDDLEvent(ddlEvent) @@ -658,7 +658,7 @@ func TestEncodeRoutedDMLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) require.NoError(t, encoder.AppendRowChangedEvent(ctx, "", rowEvent)) @@ -688,7 +688,7 @@ func TestEncodeRoutedDDLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) message, err := encoder.EncodeDDLEvent(routedDDL) @@ -711,7 +711,7 @@ func TestEncodeRoutedDDLEventUsesTargetNames(t *testing.T) { func TestEncoderOneMessage(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) helper := commonEvent.NewEventTestHelper(t) @@ -778,8 +778,15 @@ func TestEncoderMultipleMessage(t *testing.T) { `insert into test.t values (3, 333)`) ctx := context.Background() +<<<<<<< HEAD codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(400) encoder, err := NewBatchEncoder(ctx, codecConfig) +======= + codecConfig := common.NewConfig(config.ProtocolOpen). + WithMaxMessageBytes(1000). + WithMaxBatchedBytes(400) + encoder, err := NewBatchEncoder(codecConfig, nil) +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) require.NoError(t, err) insertEvents := make([]*commonEvent.RowEvent, 0, 3) @@ -856,7 +863,7 @@ func TestEncoderMultipleMessage(t *testing.T) { func TestMessageTooLarge(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(100) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) helper := commonEvent.NewEventTestHelper(t) @@ -885,6 +892,51 @@ func TestMessageTooLarge(t *testing.T) { require.Equal(t, count, 0) } +<<<<<<< HEAD +======= +func TestMessageLargerThanBatchLimit(t *testing.T) { + ctx := context.Background() + codecConfig := common.NewConfig(config.ProtocolOpen). + WithMaxMessageBytes(400). + WithMaxBatchedBytes(100) + encoder, err := NewBatchEncoder(codecConfig, nil) + require.NoError(t, err) + + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + helper.Tk().MustExec("use test") + + job := helper.DDL2Job(`create table test.t(a tinyint primary key, b int)`) + tableInfo := helper.GetTableInfo(job) + dmlEvent := helper.DML2Event("test", "t", `insert into test.t values (1, 123)`) + require.NotNil(t, dmlEvent) + insertRow, ok := dmlEvent.GetNextRow() + require.True(t, ok) + + count := 0 + insertRowEvent := &commonEvent.RowEvent{ + TableInfo: tableInfo, + CommitTs: dmlEvent.GetCommitTs(), + Event: insertRow, + ColumnSelector: columnselector.NewDefaultColumnSelector(), + Callback: func() { count += 1 }, + } + + err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) + require.NoError(t, err) + + messages := encoder.Build() + require.Len(t, messages, 1) + require.Equal(t, 1, messages[0].GetRowsCount()) + require.Greater(t, messages[0].Length(), codecConfig.MaxBatchedBytes) + require.LessOrEqual(t, messages[0].Length(), codecConfig.MaxMessageBytes) + require.Equal(t, 0, count) + + messages[0].Callback() + require.Equal(t, 1, count) +} + +>>>>>>> bc474b549 (kafka: share one claimcheck instance across encoders (#5718)) func TestLargeMessageWithHandleEnableHandleKeyOnly(t *testing.T) { helper := commonEvent.NewEventTestHelper(t) defer helper.Close() @@ -909,7 +961,7 @@ func TestLargeMessageWithHandleEnableHandleKeyOnly(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(168) codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionHandleKeyOnly - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -949,7 +1001,7 @@ func TestLargeMessageWithoutHandle(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(150) codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionHandleKeyOnly - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) helper := commonEvent.NewEventTestHelper(t) @@ -1010,7 +1062,7 @@ func TestDMLEventWithColumnSelector(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -1077,7 +1129,7 @@ func TestE2EPartitionTable(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - enc, err := NewBatchEncoder(ctx, codecConfig) + enc, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1212,7 +1264,7 @@ func TestGenerateColumn(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1347,7 +1399,7 @@ func TestDMLEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1403,7 +1455,7 @@ func TestOnlyOutputUpdatedEvent(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolOpen) codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1448,7 +1500,7 @@ func TestPKWithUK(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -1497,7 +1549,7 @@ func TestUniqueKeyWithoutPKDMLEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -1547,7 +1599,7 @@ func TestHandleOnlyEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1597,7 +1649,7 @@ func TestRenameTable(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, 0, codecConfig, nil) @@ -1655,7 +1707,7 @@ func TestDDLSequence(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolOpen) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, 0, codecConfig, nil) diff --git a/pkg/sink/codec/simple/encoder.go b/pkg/sink/codec/simple/encoder.go index b8ef228561..c8a4208f59 100644 --- a/pkg/sink/codec/simple/encoder.go +++ b/pkg/sink/codec/simple/encoder.go @@ -31,11 +31,7 @@ type Encoder struct { 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(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { marshaller, err := newMarshaller(config) if err != nil { return nil, errors.Trace(err) @@ -161,10 +157,3 @@ func (e *Encoder) EncodeDDLEvent(event *commonEvent.DDLEvent) (*common.Message, } return result, nil } - -// CleanMetrics implement the RowEventEncoderBuilder interface -func (e *Encoder) Clean() { - if e.claimCheck != nil { - e.claimCheck.CleanMetrics() - } -} diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go index 33997736af..4c5c9f5cfa 100644 --- a/pkg/sink/codec/simple/encoder_test.go +++ b/pkg/sink/codec/simple/encoder_test.go @@ -23,12 +23,14 @@ import ( "github.com/DATA-DOG/go-sqlmock" "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" + commonType "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/compression" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec/common" mock_simple "github.com/pingcap/ticdc/pkg/sink/codec/simple/mock" + "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" "github.com/pingcap/ticdc/pkg/util" timodel "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/parser/mysql" @@ -52,7 +54,7 @@ func TestEncodeCheckpoint(t *testing.T) { compression.LZ4, } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) checkpoint := 446266400629063682 @@ -97,7 +99,7 @@ func TestEncodeDMLEnableChecksum(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -147,7 +149,7 @@ func TestEncodeDMLEnableChecksum(t *testing.T) { // updateEvent.Checksum.Current = 1 // updateEvent.Checksum.Previous = 2 - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -192,7 +194,7 @@ func TestEncodeRoutedEventsUsesTargetNames(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolSimple) codecConfig.EncodingFormat = format - encIface, err := NewEncoder(ctx, codecConfig) + encIface, err := NewEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*Encoder) @@ -268,7 +270,7 @@ func TestE2EPartitionTable(t *testing.T) { common.EncodingFormatAvro, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) require.NoError(t, err) @@ -415,7 +417,7 @@ func TestEncodeDDLSequence(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -851,7 +853,7 @@ func TestEncodeDDLEvent(t *testing.T) { insertEvent.Rewind() insertEvent2.Rewind() codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -996,7 +998,7 @@ func TestColumnFlags(t *testing.T) { common.EncodingFormatJSON, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(createTableDDLEvent) @@ -1077,7 +1079,7 @@ func TestEncodeIntegerTypes(t *testing.T) { minValues.Rewind() maxValues.Rewind() codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1156,7 +1158,7 @@ func TestEncoderOtherTypes(t *testing.T) { } { event.Rewind() codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1233,7 +1235,7 @@ func TestE2EPartitionTableDMLBeforeDDL(t *testing.T) { common.EncodingFormatAvro, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) @@ -1301,7 +1303,7 @@ func TestEncodeDMLBeforeDDL(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolSimple) - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) row, ok := event.GetNextRow() @@ -1384,7 +1386,7 @@ func TestEncodeBootstrapEvent(t *testing.T) { } { dmlEvent.Rewind() codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1461,7 +1463,7 @@ func TestEncodeLargeEventsNormal(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -1543,7 +1545,7 @@ func TestDDLMessageTooLarge(t *testing.T) { common.EncodingFormatJSON, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(context.Background(), codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) _, err = enc.EncodeDDLEvent(ddlEvent) @@ -1554,6 +1556,9 @@ func TestDDLMessageTooLarge(t *testing.T) { func TestDMLMessageTooLarge(t *testing.T) { _, insertEvent, _, _ := common.NewLargeEvent4Test(t) + ctx := context.Background() + changefeedID := commonType.NewChangeFeedIDWithName("test", "") + codecConfig := common.NewConfig(config.ProtocolSimple) codecConfig.MaxMessageBytes = 50 @@ -1568,11 +1573,18 @@ func TestDMLMessageTooLarge(t *testing.T) { config.LargeMessageHandleOptionHandleKeyOnly, config.LargeMessageHandleOptionClaimCheck, } { + var ( + claimCheck *claimcheck.ClaimCheck + err error + ) codecConfig.LargeMessageHandle.LargeMessageHandleOption = handle if handle == config.LargeMessageHandleOptionClaimCheck { codecConfig.LargeMessageHandle.ClaimCheckStorageURI = "file:///tmp/simple-claim-check" + claimCheck, err = claimcheck.New(ctx, codecConfig.LargeMessageHandle, changefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) } - enc, err := NewEncoder(context.Background(), codecConfig) + enc, err := NewEncoder(codecConfig, claimCheck) require.NoError(t, err) err = enc.AppendRowChangedEvent(context.Background(), "", insertEvent) @@ -1597,6 +1609,9 @@ func TestLargerMessageHandleClaimCheck(t *testing.T) { codecConfig.LargeMessageHandle.ClaimCheckStorageURI = "file:///tmp/simple-claim-check" for _, rawValue := range []bool{false, true} { codecConfig.LargeMessageHandle.ClaimCheckRawValue = rawValue + claimCheck, err := claimcheck.New(ctx, codecConfig.LargeMessageHandle, codecConfig.ChangefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) for _, format := range []common.EncodingFormatType{ common.EncodingFormatAvro, common.EncodingFormatJSON, @@ -1610,7 +1625,7 @@ func TestLargerMessageHandleClaimCheck(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, claimCheck) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1690,7 +1705,7 @@ func TestLargeMessageHandleKeyOnly(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, db) @@ -1770,7 +1785,7 @@ func TestMarshallerError(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolSimple) - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) mockMarshaller := mock_simple.NewMockmarshaller(gomock.NewController(t)) diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index 052785e2fa..952b0a3d48 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -53,11 +53,6 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee return nil, nil } - log.Info("claim check enabled, start create the external storage", - zap.String("keyspace", changefeedID.Keyspace()), - zap.String("changefeed", changefeedID.Name()), - zap.String("storageURI", util.MaskSensitiveDataInURI(config.ClaimCheckStorageURI))) - start := time.Now() externalStorage, err := util.GetExternalStorageWithDefaultTimeout(ctx, config.ClaimCheckStorageURI) if err != nil { @@ -70,12 +65,6 @@ func New(ctx context.Context, config *config.LargeMessageHandleConfig, changefee return nil, errors.Trace(err) } - log.Info("claim-check create the external storage success", - zap.String("keyspace", changefeedID.Keyspace()), - zap.String("changefeed", changefeedID.Name()), - zap.String("storageURI", util.MaskSensitiveDataInURI(config.ClaimCheckStorageURI)), - zap.Duration("duration", time.Since(start))) - return &ClaimCheck{ changefeedID: changefeedID, storage: externalStorage, @@ -112,8 +101,15 @@ func (c *ClaimCheck) FileNameWithPrefix(fileName string) string { return strings.TrimSuffix(c.storage.URI(), "/") + "/" + fileName } -// CleanMetrics the claim check by clean up the metrics. -func (c *ClaimCheck) CleanMetrics() { +// Close closes the claim-check storage. +func (c *ClaimCheck) Close() { + if c == nil { + return + } + + if c.storage != nil { + c.storage.Close() + } claimCheckSendMessageDuration.DeleteLabelValues(c.changefeedID.Keyspace(), c.changefeedID.Name()) claimCheckSendMessageCount.DeleteLabelValues(c.changefeedID.Keyspace(), c.changefeedID.Name()) } diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go new file mode 100644 index 0000000000..e51cfc5626 --- /dev/null +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -0,0 +1,96 @@ +// Copyright 2023 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package 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) { + t.Parallel() + + ctx := context.Background() + + changefeedID := commonType.NewChangeFeedIDWithName("test", "") + largeHandleConfig := config.NewDefaultLargeMessageHandleConfig() + + claimCheck, err := New(ctx, largeHandleConfig, changefeedID) + require.NoError(t, err) + require.Nil(t, claimCheck) + + largeHandleConfig.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck + largeHandleConfig.ClaimCheckStorageURI = "file:///tmp/abc/" + claimCheck, err = New(ctx, largeHandleConfig, changefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) + + fileName := claimCheck.FileNameWithPrefix("file.json") + require.Equal(t, "file:///tmp/abc/file.json", fileName) +} + +func TestClaimCheckCloseClosesStorage(t *testing.T) { + var nilClaimCheck *ClaimCheck + require.NotPanics(t, nilClaimCheck.Close) + + ctrl := gomock.NewController(t) + storage := mockobjstore.NewMockStorage(ctrl) + storage.EXPECT().Close().Times(1) + claimCheck := &ClaimCheck{ + storage: storage, + changefeedID: commonType.NewChangeFeedIDWithName("test", "default"), + } + + 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)) + } +}