diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index bd776c5e49..7d2e78ca54 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 newKafkaSinkComponent( @@ -55,38 +60,47 @@ func newKafkaSinkComponent( sinkURI *url.URL, sinkConfig *config.SinkConfig, ) (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 - kafkaComponent.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) + comp.factory, err = kafka.NewSaramaFactory(ctx, options, changefeedID) if err != nil { - return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) } isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro - kafkaComponent.eventRouter, err = eventrouter.NewEventRouter( + comp.eventRouter, err = eventrouter.NewEventRouter( sinkConfig, topic, false, isAvroLike) 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( @@ -94,41 +108,38 @@ func newKafkaSinkComponent( options.MaxMessageBytes, options.MaxBatchedBytes, ) 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 } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 9de1070e15..8cd9ed3be3 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -30,6 +30,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/ticdc/pkg/util" "github.com/pingcap/ticdc/utils/chann" "go.uber.org/atomic" @@ -95,6 +96,12 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. 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) @@ -138,12 +145,10 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.WrapError(errors.ErrKafkaCreateTopic, err) } - encoder, err := codec.NewEventEncoder(ctx, encoderConfig) + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { return errors.Trace(err) } - encoder.Clean() - return nil } @@ -164,24 +169,26 @@ 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() }() - statistics := metrics.NewStatistics(changefeedID, keyspaceID, "sink") 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 6d7d337574..bfed3536c0 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -117,11 +117,11 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, if err != nil { return nil, err } - encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) + encoderGroup, err := codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, nil, changefeedID) if err != nil { return nil, err } - encoder, err := codec.NewEventEncoder(ctx, encoderConfig) + encoder, err := codec.NewEventEncoder(ctx, encoderConfig, nil) if err != nil { return nil, err } diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index 3ccb299767..12e1beda52 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -130,12 +130,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 223af820ce..976b795d0b 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 e6415d06f6..b0c3c230e2 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -27,23 +27,24 @@ 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.ProtocolDebeziumAvro: return debezium.NewAvroBatchEncoder(ctx, cfg, config.GetGlobalServerConfig().ClusterID) 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 65191626bc..b0075940c9 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 c92948d198..4931fdd515 100644 --- a/pkg/sink/codec/canal/canal_json_test.go +++ b/pkg/sink/codec/canal/canal_json_test.go @@ -25,6 +25,7 @@ import ( "github.com/pingcap/ticdc/pkg/config/kerneltype" "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" ) @@ -68,7 +69,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) @@ -152,7 +153,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) @@ -214,7 +215,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) @@ -263,7 +264,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) @@ -313,7 +314,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) @@ -363,7 +364,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) @@ -413,7 +414,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) @@ -472,7 +473,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) @@ -530,7 +531,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) @@ -589,7 +590,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) @@ -675,7 +676,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) @@ -773,7 +774,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) @@ -864,7 +868,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) @@ -950,7 +954,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) @@ -982,7 +986,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) @@ -1013,7 +1017,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) @@ -1285,7 +1289,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) @@ -1316,7 +1320,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) @@ -1326,7 +1330,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) @@ -1386,7 +1390,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 85621f230a..7202965d97 100644 --- a/pkg/sink/codec/debezium/encoder.go +++ b/pkg/sink/codec/debezium/encoder.go @@ -201,8 +201,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 17a323cb7d..b8e608a298 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 d637af72ee..de88347061 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) @@ -781,7 +781,7 @@ func TestEncoderMultipleMessage(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolOpen). WithMaxMessageBytes(1000). WithMaxBatchedBytes(400) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) insertEvents := make([]*commonEvent.RowEvent, 0, 3) @@ -860,7 +860,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) @@ -894,7 +894,7 @@ func TestMessageLargerThanBatchLimit(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolOpen). WithMaxMessageBytes(400). WithMaxBatchedBytes(100) - encoder, err := NewBatchEncoder(ctx, codecConfig) + encoder, err := NewBatchEncoder(codecConfig, nil) require.NoError(t, err) helper := commonEvent.NewEventTestHelper(t) @@ -955,7 +955,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) @@ -995,7 +995,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) @@ -1056,7 +1056,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) @@ -1123,7 +1123,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) @@ -1258,7 +1258,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) @@ -1391,7 +1391,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) @@ -1447,7 +1447,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) @@ -1492,7 +1492,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) @@ -1541,7 +1541,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) @@ -1591,7 +1591,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) @@ -1641,7 +1641,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) @@ -1699,7 +1699,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 bea43f4102..dc2a7456a4 100644 --- a/pkg/sink/codec/simple/encoder_test.go +++ b/pkg/sink/codec/simple/encoder_test.go @@ -23,6 +23,7 @@ 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" @@ -30,6 +31,7 @@ import ( "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" ticonfig "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/dxf/framework/handle" @@ -63,7 +65,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 @@ -108,7 +110,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) @@ -158,7 +160,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) @@ -203,7 +205,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) @@ -279,7 +281,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) @@ -426,7 +428,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) @@ -862,7 +864,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) @@ -1007,7 +1009,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) @@ -1088,7 +1090,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) @@ -1167,7 +1169,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) @@ -1244,7 +1246,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) @@ -1313,7 +1315,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() @@ -1397,7 +1399,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) @@ -1474,7 +1476,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) @@ -1556,7 +1558,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) @@ -1567,6 +1569,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 @@ -1581,11 +1586,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) @@ -1610,6 +1622,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, @@ -1623,7 +1638,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) @@ -1703,7 +1718,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) @@ -1784,7 +1799,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 1b25528be2..e38b01c97f 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -49,11 +49,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 { @@ -66,12 +61,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, @@ -108,8 +97,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 index a641ddd97b..e51cfc5626 100644 --- a/pkg/sink/kafka/claimcheck/claim_check_test.go +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -15,11 +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) { @@ -38,7 +43,54 @@ func TestClaimCheck(t *testing.T) { 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)) + } +}