From e81d21697854605ca17dc406cc9b54198714bb00 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 15:04:22 +0800 Subject: [PATCH 1/8] pass claim check into the encoder --- pkg/sink/codec/builder.go | 12 +++- pkg/sink/codec/canal/canal_json_encoder.go | 6 +- .../codec/canal/canal_json_encoder_test.go | 38 ++++++------ pkg/sink/codec/canal/canal_json_test.go | 40 ++++++------- pkg/sink/codec/internal/claimcheck/context.go | 31 ++++++++++ pkg/sink/codec/open/encoder.go | 6 +- pkg/sink/codec/open/encoder_test.go | 60 +++++++++---------- pkg/sink/codec/simple/encoder.go | 6 +- pkg/sink/codec/simple/encoder_test.go | 49 +++++++++------ pkg/sink/kafka/claimcheck/claim_check.go | 11 ---- 10 files changed, 142 insertions(+), 117 deletions(-) create mode 100644 pkg/sink/codec/internal/claimcheck/context.go diff --git a/pkg/sink/codec/builder.go b/pkg/sink/codec/builder.go index e6415d06f6..f8a1f81b4a 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -27,23 +27,29 @@ 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) { + claimCheck, err := claimcheck.New(ctx, cfg.LargeMessageHandle, cfg.ChangefeedID) + if err != nil { + return nil, errors.Trace(err) + } + switch cfg.Protocol { case config.ProtocolDefault, config.ProtocolOpen: - return open.NewBatchEncoder(ctx, cfg) + return open.NewBatchEncoder(ctx, cfg, claimCheck) case config.ProtocolAvro: return avro.NewAvroEncoder(ctx, cfg) case config.ProtocolCanalJSON: - return canal.NewJSONRowEventEncoder(ctx, cfg) + return canal.NewJSONRowEventEncoder(ctx, 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(ctx, 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..9db231d7cd 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(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { return &JSONRowEventEncoder{ messages: make([]*common.Message, 0, 1), config: config, diff --git a/pkg/sink/codec/canal/canal_json_encoder_test.go b/pkg/sink/codec/canal/canal_json_encoder_test.go index 65191626bc..232a1bcc0f 100644 --- a/pkg/sink/codec/canal/canal_json_encoder_test.go +++ b/pkg/sink/codec/canal/canal_json_encoder_test.go @@ -47,7 +47,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(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -131,7 +131,7 @@ func TestCanalJSONCompressionE2E(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4 ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -208,7 +208,7 @@ func TestEncodeRoutedDMLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -237,7 +237,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(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -270,7 +270,7 @@ func TestCanalJSONClaimCheckE2E(t *testing.T) { for _, rawValue := range []bool{false, true} { codecConfig.LargeMessageHandle.ClaimCheckRawValue = rawValue - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -317,7 +317,7 @@ func TestNewCanalJSONMessageHandleKeyOnly4LargeMessage(t *testing.T) { ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -362,7 +362,7 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -382,7 +382,7 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { require.Equal(t, "CREATE", msg.EventType) codecConfig.EnableTiDBExtension = true - encIface, err = NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -399,7 +399,7 @@ 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(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) require.NotNil(t, encoder) @@ -440,7 +440,7 @@ func TestEncodeCheckpointEvent(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = enable - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) msg, err := encoder.EncodeCheckpointEvent(watermark) @@ -484,7 +484,7 @@ func TestCheckpointEventValueMarshal(t *testing.T) { ctx := context.Background() - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) var watermark uint64 = 1024 @@ -519,7 +519,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(ctx, codecConfig, nil) require.NoError(t, err) require.NotNil(t, encoder) @@ -563,7 +563,7 @@ func TestCanalJSONAppendRowChangedEventWithCallback(t *testing.T) { codecConfig.EnableTiDBExtension = true ctx := context.Background() - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) count := 0 @@ -654,7 +654,7 @@ func TestMaxMessageBytes(t *testing.T) { maxMessageBytes := 300 codecConfig := common.NewConfig(config.ProtocolCanalJSON).WithMaxMessageBytes(maxMessageBytes) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -669,7 +669,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(ctx, codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -689,7 +689,7 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -737,7 +737,7 @@ func TestE2EPartitionTableByHash(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -794,7 +794,7 @@ func TestE2EPartitionTableByRange(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -858,7 +858,7 @@ func TestE2EPartitionTable(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, 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..7a26f8b392 100644 --- a/pkg/sink/codec/canal/canal_json_test.go +++ b/pkg/sink/codec/canal/canal_json_test.go @@ -68,7 +68,7 @@ func TestIntegerContentCompatible(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -152,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(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) @@ -214,7 +214,7 @@ func TestFloatTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -263,7 +263,7 @@ func TestTimeTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -313,7 +313,7 @@ func TestStringTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -363,7 +363,7 @@ func TestBlobTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -413,7 +413,7 @@ func TestTextTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -472,7 +472,7 @@ func TestOtherTypes(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -530,7 +530,7 @@ func TestDMLEventWithColumnSelector(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -589,7 +589,7 @@ func TestDMLMultiplePK(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -675,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(context.Background(), codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(context.Background(), "", rowEvent) require.ErrorIs(t, err, errors.ErrMessageTooLarge) @@ -773,7 +773,7 @@ func TestLargeMessageClaimCheck(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = "snappy" codecConfig.LargeMessageHandle.ClaimCheckStorageURI = "file:///tmp/canal-json-claim-check" - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertEvent) @@ -864,7 +864,7 @@ func TestMessageLargeHandleKeyOnly(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionHandleKeyOnly codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -950,7 +950,7 @@ func TestDMLTypeEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -982,7 +982,7 @@ func TestDMLTypeEvent(t *testing.T) { // update with only updated columns codecConfig.OnlyOutputUpdatedColumns = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", updateEvent) @@ -1013,7 +1013,7 @@ func TestDDLSequence(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) @@ -1285,7 +1285,7 @@ func TestCreateTableDDL(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) message, err := encoder.EncodeDDLEvent(ddlEvent) @@ -1316,7 +1316,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(ctx, codecConfig, nil) require.NoError(t, err) watermark := uint64(179394) @@ -1326,7 +1326,7 @@ func TestCheckpointTs(t *testing.T) { // with extension codecConfig.EnableTiDBExtension = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) message, err = encoder.EncodeCheckpointEvent(watermark) require.NoError(t, err) @@ -1386,7 +1386,7 @@ func TestRowKey(t *testing.T) { codecConfig.OnlyOutputUpdatedColumns = true codecConfig.EnableTiDBExtension = true codecConfig.OutputRowKey = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) require.NoError(t, err) diff --git a/pkg/sink/codec/internal/claimcheck/context.go b/pkg/sink/codec/internal/claimcheck/context.go new file mode 100644 index 0000000000..a339d67d07 --- /dev/null +++ b/pkg/sink/codec/internal/claimcheck/context.go @@ -0,0 +1,31 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package claimcheck + +import ( + "context" + + kafkaclaimcheck "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" +) + +type contextKey struct{} + +func Bind(ctx context.Context, claimCheck *kafkaclaimcheck.ClaimCheck) context.Context { + return context.WithValue(ctx, contextKey{}, claimCheck) +} + +func From(ctx context.Context) *kafkaclaimcheck.ClaimCheck { + claimCheck, _ := ctx.Value(contextKey{}).(*kafkaclaimcheck.ClaimCheck) + return claimCheck +} diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index 17a323cb7d..b96852da2b 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(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { lock.Lock() clear(columnFlagsCache) lock.Unlock() diff --git a/pkg/sink/codec/open/encoder_test.go b/pkg/sink/codec/open/encoder_test.go index d637af72ee..1ef565b607 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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..c90b1d519c 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(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { marshaller, err := newMarshaller(config) if err != nil { return nil, errors.Trace(err) diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go index bea43f4102..03ff2ba6fb 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(ctx, 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(context.Background(), 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,17 @@ 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) } - enc, err := NewEncoder(context.Background(), codecConfig) + enc, err := NewEncoder(context.Background(), codecConfig, claimCheck) require.NoError(t, err) err = enc.AppendRowChangedEvent(context.Background(), "", insertEvent) @@ -1623,7 +1634,7 @@ func TestLargerMessageHandleClaimCheck(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(ctx, codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1703,7 +1714,7 @@ func TestLargeMessageHandleKeyOnly(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(ctx, codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, db) @@ -1784,7 +1795,7 @@ func TestMarshallerError(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolSimple) - enc, err := NewEncoder(ctx, codecConfig) + enc, err := NewEncoder(ctx, 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..f3f61321c9 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, From 14db0d3dfe014591f085f32ac4b3349e50d4200c Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 15:11:28 +0800 Subject: [PATCH 2/8] share claim check among all encoders --- downstreamadapter/sink/kafka/helper.go | 2 +- downstreamadapter/sink/kafka/sink_test.go | 2 +- downstreamadapter/sink/pulsar/helper.go | 2 +- pkg/sink/codec/builder.go | 7 +------ pkg/sink/codec/encoder_group.go | 12 +++++++++--- 5 files changed, 13 insertions(+), 12 deletions(-) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index bd776c5e49..941642bcc8 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -102,7 +102,7 @@ func newKafkaSinkComponent( return kafkaComponent, protocol, errors.Trace(err) } - kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig) + kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, nil) if err != nil { return kafkaComponent, protocol, errors.Trace(err) } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 6d7d337574..2b8316a59b 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -121,7 +121,7 @@ func newKafkaSinkForTestWithProducers(ctx context.Context, 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..e0727f08e5 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -135,7 +135,7 @@ func newPulsarSinkComponentWithFactory(ctx context.Context, 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/builder.go b/pkg/sink/codec/builder.go index f8a1f81b4a..29e69218eb 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -31,12 +31,7 @@ import ( "go.uber.org/zap" ) -func NewEventEncoder(ctx context.Context, cfg *common.Config) (common.EventEncoder, error) { - claimCheck, err := claimcheck.New(ctx, cfg.LargeMessageHandle, cfg.ChangefeedID) - if err != nil { - return nil, errors.Trace(err) - } - +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, claimCheck) diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go index 7ae503c985..1ec02dc3c5 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" @@ -73,12 +74,17 @@ func NewEncoderGroup( if concurrency <= 0 { concurrency = config.DefaultEncoderGroupConcurrency } + + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return nil, errors.Trace(err) + } + 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 +94,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) From 75cdb220e834aa0d4124d889bc3dbf17bc311d8f Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 15:39:02 +0800 Subject: [PATCH 3/8] remove the clean from the encoder interface --- downstreamadapter/sink/kafka/sink.go | 4 +- pkg/sink/codec/avro/arvo.go | 2 - pkg/sink/codec/bootstraper.go | 1 - pkg/sink/codec/canal/canal_json_encoder.go | 6 --- pkg/sink/codec/common/encoder.go | 2 - pkg/sink/codec/debezium/encoder.go | 2 - pkg/sink/codec/encoder_group.go | 3 -- pkg/sink/codec/open/encoder.go | 6 --- pkg/sink/codec/simple/encoder.go | 7 --- pkg/sink/kafka/claimcheck/claim_check.go | 6 ++- pkg/sink/kafka/claimcheck/claim_check_test.go | 48 +++++++++++++++++++ 11 files changed, 53 insertions(+), 34 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 9de1070e15..28faa74b3d 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -138,12 +138,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, nil) if err != nil { return errors.Trace(err) } - encoder.Clean() - return nil } 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/canal/canal_json_encoder.go b/pkg/sink/codec/canal/canal_json_encoder.go index 9db231d7cd..34a71731ad 100644 --- a/pkg/sink/codec/canal/canal_json_encoder.go +++ b/pkg/sink/codec/canal/canal_json_encoder.go @@ -578,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/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 1ec02dc3c5..756c6d0cf9 100644 --- a/pkg/sink/codec/encoder_group.go +++ b/pkg/sink/codec/encoder_group.go @@ -212,9 +212,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 b96852da2b..39523ffb07 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -60,12 +60,6 @@ func NewBatchEncoder(ctx context.Context, config *common.Config, claimCheck *cla }, 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/simple/encoder.go b/pkg/sink/codec/simple/encoder.go index c90b1d519c..2d28e82ec4 100644 --- a/pkg/sink/codec/simple/encoder.go +++ b/pkg/sink/codec/simple/encoder.go @@ -157,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/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index f3f61321c9..6f2520a9e7 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -97,8 +97,10 @@ 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() { +func (c *ClaimCheck) Close() { + 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..b3e0bee671 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) { @@ -42,3 +47,46 @@ func TestClaimCheck(t *testing.T) { fileName := claimCheck.FileNameWithPrefix("file.json") require.Equal(t, "file:///tmp/abc/file.json", fileName) } + +func TestClaimCheckCloseClosesStorage(t *testing.T) { + ctrl := gomock.NewController(t) + storage := mockobjstore.NewMockStorage(ctrl) + storage.EXPECT().Close().Times(1) + claimCheck := &ClaimCheck{ + storage: storage, + changefeedID: commonType.NewChangeFeedIDWithName("test", "default"), + } + + claimCheck.Close() +} + +func TestClaimCheckConcurrentWrites(t *testing.T) { + ctx := context.Background() + storage := objstore.NewMemStorage() + changefeedID := commonType.NewChangeFeedIDWithName("test", "default") + claimCheck := &ClaimCheck{ + storage: storage, + rawValue: true, + changefeedID: changefeedID, + metricSendMessageDuration: claimCheckSendMessageDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + metricSendMessageCount: claimCheckSendMessageCount.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()), + } + t.Cleanup(claimCheck.Close) + + const concurrency = 32 + group := new(errgroup.Group) + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + group.Go(func() error { + return claimCheck.WriteMessage(ctx, nil, []byte(fileName), fileName) + }) + } + require.NoError(t, group.Wait()) + + for i := range concurrency { + fileName := fmt.Sprintf("%d.json", i) + data, err := storage.ReadFile(ctx, fileName) + require.NoError(t, err) + require.Equal(t, fileName, string(data)) + } +} From 71d9682f51bc28016d34ebcf60c029c23c4a0a41 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 15:48:39 +0800 Subject: [PATCH 4/8] claimcheck at the kafka sink level --- downstreamadapter/sink/kafka/helper.go | 20 ++++++++++++++++++-- downstreamadapter/sink/kafka/sink.go | 9 ++++++++- downstreamadapter/sink/kafka/sink_test.go | 2 +- downstreamadapter/sink/pulsar/helper.go | 2 +- pkg/sink/codec/encoder_group.go | 8 +++----- 5 files changed, 31 insertions(+), 10 deletions(-) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index 941642bcc8..98c299bd4a 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( @@ -97,12 +102,23 @@ func newKafkaSinkComponent( return kafkaComponent, protocol, errors.Trace(err) } - kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, changefeedID) + claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) + if err != nil { + return kafkaComponent, protocol, errors.Trace(err) + } + defer func() { + if err != nil { + claimCheck.Close() + } + }() + kafkaComponent.claimCheck = claimCheck + + kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, claimCheck, changefeedID) if err != nil { return kafkaComponent, protocol, errors.Trace(err) } - kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, nil) + kafkaComponent.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { return kafkaComponent, protocol, errors.Trace(err) } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 28faa74b3d..787647527a 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,7 +145,7 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url. return errors.WrapError(errors.ErrKafkaCreateTopic, err) } - _, err = codec.NewEventEncoder(ctx, encoderConfig, nil) + _, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) if err != nil { return errors.Trace(err) } diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go index 2b8316a59b..bfed3536c0 100644 --- a/downstreamadapter/sink/kafka/sink_test.go +++ b/downstreamadapter/sink/kafka/sink_test.go @@ -117,7 +117,7 @@ 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 } diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go index e0727f08e5..12e1beda52 100644 --- a/downstreamadapter/sink/pulsar/helper.go +++ b/downstreamadapter/sink/pulsar/helper.go @@ -130,7 +130,7 @@ 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) } diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go index 756c6d0cf9..eeeca21a20 100644 --- a/pkg/sink/codec/encoder_group.go +++ b/pkg/sink/codec/encoder_group.go @@ -68,6 +68,7 @@ func NewEncoderGroup( ctx context.Context, cfg *config.SinkConfig, encoderConfig *common.Config, + claimCheck *claimcheck.ClaimCheck, changefeedID commonType.ChangeFeedID, ) (*encoderGroup, error) { concurrency := util.GetOrZero(cfg.EncoderConcurrency) @@ -75,13 +76,10 @@ func NewEncoderGroup( concurrency = config.DefaultEncoderGroupConcurrency } - claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID) - if err != nil { - return nil, errors.Trace(err) - } - 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, claimCheck) From 53384e05faa2809301162cfb8dc5725bde826ae7 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 15:51:57 +0800 Subject: [PATCH 5/8] remove it --- pkg/sink/codec/internal/claimcheck/context.go | 31 ------------------- 1 file changed, 31 deletions(-) delete mode 100644 pkg/sink/codec/internal/claimcheck/context.go diff --git a/pkg/sink/codec/internal/claimcheck/context.go b/pkg/sink/codec/internal/claimcheck/context.go deleted file mode 100644 index a339d67d07..0000000000 --- a/pkg/sink/codec/internal/claimcheck/context.go +++ /dev/null @@ -1,31 +0,0 @@ -// Copyright 2026 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package claimcheck - -import ( - "context" - - kafkaclaimcheck "github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck" -) - -type contextKey struct{} - -func Bind(ctx context.Context, claimCheck *kafkaclaimcheck.ClaimCheck) context.Context { - return context.WithValue(ctx, contextKey{}, claimCheck) -} - -func From(ctx context.Context) *kafkaclaimcheck.ClaimCheck { - claimCheck, _ := ctx.Value(contextKey{}).(*kafkaclaimcheck.ClaimCheck) - return claimCheck -} From ee0f0d662498ec31517833822bc510f514cdfa33 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 17:16:39 +0800 Subject: [PATCH 6/8] simplify the code --- downstreamadapter/sink/kafka/sink.go | 20 ++++++++++--------- .../codec/canal/canal_json_encoder_test.go | 6 +++++- pkg/sink/codec/canal/canal_json_test.go | 6 +++++- pkg/sink/codec/simple/encoder_test.go | 6 +++++- pkg/sink/kafka/claimcheck/claim_check.go | 5 +++++ pkg/sink/kafka/claimcheck/claim_check_test.go | 4 ++++ 6 files changed, 35 insertions(+), 12 deletions(-) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 787647527a..8cd9ed3be3 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -169,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/pkg/sink/codec/canal/canal_json_encoder_test.go b/pkg/sink/codec/canal/canal_json_encoder_test.go index 232a1bcc0f..b4f655c7e7 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" ) @@ -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, nil) + encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, claimCheck) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) diff --git a/pkg/sink/codec/canal/canal_json_test.go b/pkg/sink/codec/canal/canal_json_test.go index 7a26f8b392..4561a04b43 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" ) @@ -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, nil) + claimCheck, err := claimcheck.New(ctx, codecConfig.LargeMessageHandle, codecConfig.ChangefeedID) + require.NoError(t, err) + t.Cleanup(claimCheck.Close) + encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, claimCheck) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertEvent) diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go index 03ff2ba6fb..104c9139b0 100644 --- a/pkg/sink/codec/simple/encoder_test.go +++ b/pkg/sink/codec/simple/encoder_test.go @@ -1595,6 +1595,7 @@ func TestDMLMessageTooLarge(t *testing.T) { 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, claimCheck) require.NoError(t, err) @@ -1621,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, @@ -1634,7 +1638,7 @@ func TestLargerMessageHandleClaimCheck(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(ctx, codecConfig, claimCheck) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go index 6f2520a9e7..e38b01c97f 100644 --- a/pkg/sink/kafka/claimcheck/claim_check.go +++ b/pkg/sink/kafka/claimcheck/claim_check.go @@ -97,7 +97,12 @@ func (c *ClaimCheck) FileNameWithPrefix(fileName string) string { return strings.TrimSuffix(c.storage.URI(), "/") + "/" + fileName } +// Close closes the claim-check storage. func (c *ClaimCheck) Close() { + if c == nil { + return + } + if c.storage != nil { c.storage.Close() } diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go index b3e0bee671..e51cfc5626 100644 --- a/pkg/sink/kafka/claimcheck/claim_check_test.go +++ b/pkg/sink/kafka/claimcheck/claim_check_test.go @@ -43,12 +43,16 @@ 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) From c603b163d6bbf2ef67e2e5559169e95b6bc759eb Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 17:44:42 +0800 Subject: [PATCH 7/8] simplify the code --- pkg/sink/codec/builder.go | 6 +- pkg/sink/codec/canal/canal_json_encoder.go | 2 +- .../codec/canal/canal_json_encoder_test.go | 46 ++++++-------- pkg/sink/codec/canal/canal_json_test.go | 40 ++++++------- pkg/sink/codec/open/encoder.go | 2 +- pkg/sink/codec/open/encoder_test.go | 60 +++++++++---------- pkg/sink/codec/simple/encoder.go | 2 +- pkg/sink/codec/simple/encoder_test.go | 38 ++++++------ 8 files changed, 94 insertions(+), 102 deletions(-) diff --git a/pkg/sink/codec/builder.go b/pkg/sink/codec/builder.go index 29e69218eb..b0c3c230e2 100644 --- a/pkg/sink/codec/builder.go +++ b/pkg/sink/codec/builder.go @@ -34,17 +34,17 @@ import ( 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, claimCheck) + return open.NewBatchEncoder(cfg, claimCheck) case config.ProtocolAvro: return avro.NewAvroEncoder(ctx, cfg) case config.ProtocolCanalJSON: - return canal.NewJSONRowEventEncoder(ctx, cfg, claimCheck) + 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, claimCheck) + 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 34a71731ad..dc076f3ee0 100644 --- a/pkg/sink/codec/canal/canal_json_encoder.go +++ b/pkg/sink/codec/canal/canal_json_encoder.go @@ -373,7 +373,7 @@ type JSONRowEventEncoder struct { } // NewJSONRowEventEncoder creates a new JSONRowEventEncoder -func NewJSONRowEventEncoder(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { +func NewJSONRowEventEncoder(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { return &JSONRowEventEncoder{ messages: make([]*common.Message, 0, 1), config: config, diff --git a/pkg/sink/codec/canal/canal_json_encoder_test.go b/pkg/sink/codec/canal/canal_json_encoder_test.go index b4f655c7e7..b0075940c9 100644 --- a/pkg/sink/codec/canal/canal_json_encoder_test.go +++ b/pkg/sink/codec/canal/canal_json_encoder_test.go @@ -48,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, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -132,7 +132,7 @@ func TestCanalJSONCompressionE2E(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4 ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -209,7 +209,7 @@ func TestEncodeRoutedDMLEventUsesTargetNames(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -238,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, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -274,7 +274,7 @@ func TestCanalJSONClaimCheckE2E(t *testing.T) { require.NoError(t, err) t.Cleanup(claimCheck.Close) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, claimCheck) + encIface, err := NewJSONRowEventEncoder(codecConfig, claimCheck) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -319,9 +319,7 @@ func TestNewCanalJSONMessageHandleKeyOnly4LargeMessage(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4 codecConfig.MaxMessageBytes = 500 - ctx := context.Background() - - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -364,9 +362,8 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { defer helper.Close() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - ctx := context.Background() - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -386,7 +383,7 @@ func TestNewCanalJSONMessageFromDDL(t *testing.T) { require.Equal(t, "CREATE", msg.EventType) codecConfig.EnableTiDBExtension = true - encIface, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -401,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, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) require.NotNil(t, encoder) @@ -438,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, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) msg, err := encoder.EncodeCheckpointEvent(watermark) @@ -486,9 +481,7 @@ func TestCheckpointEventValueMarshal(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - ctx := context.Background() - - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) var watermark uint64 = 1024 @@ -523,7 +516,7 @@ func TestDDLEventWithExtension(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) require.NotNil(t, encoder) @@ -565,9 +558,8 @@ func TestCanalJSONAppendRowChangedEventWithCallback(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.EnableTiDBExtension = true - ctx := context.Background() - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) count := 0 @@ -658,7 +650,7 @@ func TestMaxMessageBytes(t *testing.T) { maxMessageBytes := 300 codecConfig := common.NewConfig(config.ProtocolCanalJSON).WithMaxMessageBytes(maxMessageBytes) - encIface, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encIface, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*JSONRowEventEncoder) @@ -673,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, nil) + encIface, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) encoder = encIface.(*JSONRowEventEncoder) @@ -693,7 +685,7 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -741,7 +733,7 @@ func TestE2EPartitionTableByHash(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -798,7 +790,7 @@ func TestE2EPartitionTableByRange(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -862,7 +854,7 @@ func TestE2EPartitionTable(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + 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 4561a04b43..4931fdd515 100644 --- a/pkg/sink/codec/canal/canal_json_test.go +++ b/pkg/sink/codec/canal/canal_json_test.go @@ -69,7 +69,7 @@ func TestIntegerContentCompatible(t *testing.T) { codecConfig.ContentCompatible = true codecConfig.OnlyOutputUpdatedColumns = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent) @@ -153,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, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) @@ -215,7 +215,7 @@ func TestFloatTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -264,7 +264,7 @@ func TestTimeTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -314,7 +314,7 @@ func TestStringTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -364,7 +364,7 @@ func TestBlobTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -414,7 +414,7 @@ func TestTextTypes(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -473,7 +473,7 @@ func TestOtherTypes(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -531,7 +531,7 @@ func TestDMLEventWithColumnSelector(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -590,7 +590,7 @@ func TestDMLMultiplePK(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolCanalJSON) codecConfig.ContentCompatible = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -676,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, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(context.Background(), "", rowEvent) require.ErrorIs(t, err, errors.ErrMessageTooLarge) @@ -777,7 +777,7 @@ func TestLargeMessageClaimCheck(t *testing.T) { claimCheck, err := claimcheck.New(ctx, codecConfig.LargeMessageHandle, codecConfig.ChangefeedID) require.NoError(t, err) t.Cleanup(claimCheck.Close) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, claimCheck) + encoder, err := NewJSONRowEventEncoder(codecConfig, claimCheck) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", insertEvent) @@ -868,7 +868,7 @@ func TestMessageLargeHandleKeyOnly(t *testing.T) { codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionHandleKeyOnly codecConfig.EnableTiDBExtension = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", rowEvent) @@ -954,7 +954,7 @@ func TestDMLTypeEvent(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) decoder, err := NewDecoder(ctx, codecConfig, nil) @@ -986,7 +986,7 @@ func TestDMLTypeEvent(t *testing.T) { // update with only updated columns codecConfig.OnlyOutputUpdatedColumns = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", updateEvent) @@ -1017,7 +1017,7 @@ func TestDDLSequence(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolCanalJSON) - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) @@ -1289,7 +1289,7 @@ func TestCreateTableDDL(t *testing.T) { for _, enableTiDBExtension := range []bool{false, true} { codecConfig.EnableTiDBExtension = enableTiDBExtension - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) message, err := encoder.EncodeDDLEvent(ddlEvent) @@ -1320,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, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) watermark := uint64(179394) @@ -1330,7 +1330,7 @@ func TestCheckpointTs(t *testing.T) { // with extension codecConfig.EnableTiDBExtension = true - encoder, err = NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err = NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) message, err = encoder.EncodeCheckpointEvent(watermark) require.NoError(t, err) @@ -1390,7 +1390,7 @@ func TestRowKey(t *testing.T) { codecConfig.OnlyOutputUpdatedColumns = true codecConfig.EnableTiDBExtension = true codecConfig.OutputRowKey = true - encoder, err := NewJSONRowEventEncoder(ctx, codecConfig, nil) + encoder, err := NewJSONRowEventEncoder(codecConfig, nil) require.NoError(t, err) err = encoder.AppendRowChangedEvent(ctx, "", event) require.NoError(t, err) diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go index 39523ffb07..b8e608a298 100644 --- a/pkg/sink/codec/open/encoder.go +++ b/pkg/sink/codec/open/encoder.go @@ -50,7 +50,7 @@ type batchEncoder struct { } // NewBatchEncoder creates a new batchEncoder. -func NewBatchEncoder(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { +func NewBatchEncoder(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { lock.Lock() clear(columnFlagsCache) lock.Unlock() diff --git a/pkg/sink/codec/open/encoder_test.go b/pkg/sink/codec/open/encoder_test.go index 1ef565b607..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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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, nil) + 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 2d28e82ec4..c8a4208f59 100644 --- a/pkg/sink/codec/simple/encoder.go +++ b/pkg/sink/codec/simple/encoder.go @@ -31,7 +31,7 @@ type Encoder struct { marshaller marshaller } -func NewEncoder(ctx context.Context, config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { +func NewEncoder(config *common.Config, claimCheck *claimcheck.ClaimCheck) (common.EventEncoder, error) { marshaller, err := newMarshaller(config) if err != nil { return nil, errors.Trace(err) diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go index 104c9139b0..dc2a7456a4 100644 --- a/pkg/sink/codec/simple/encoder_test.go +++ b/pkg/sink/codec/simple/encoder_test.go @@ -65,7 +65,7 @@ func TestEncodeCheckpoint(t *testing.T) { compression.LZ4, } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) checkpoint := 446266400629063682 @@ -110,7 +110,7 @@ func TestEncodeDMLEnableChecksum(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -160,7 +160,7 @@ func TestEncodeDMLEnableChecksum(t *testing.T) { // updateEvent.Checksum.Current = 1 // updateEvent.Checksum.Previous = 2 - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -205,7 +205,7 @@ func TestEncodeRoutedEventsUsesTargetNames(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolSimple) codecConfig.EncodingFormat = format - encIface, err := NewEncoder(ctx, codecConfig, nil) + encIface, err := NewEncoder(codecConfig, nil) require.NoError(t, err) encoder := encIface.(*Encoder) @@ -281,7 +281,7 @@ func TestE2EPartitionTable(t *testing.T) { common.EncodingFormatAvro, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) require.NoError(t, err) @@ -428,7 +428,7 @@ func TestEncodeDDLSequence(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -864,7 +864,7 @@ func TestEncodeDDLEvent(t *testing.T) { insertEvent.Rewind() insertEvent2.Rewind() codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -1009,7 +1009,7 @@ func TestColumnFlags(t *testing.T) { common.EncodingFormatJSON, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(createTableDDLEvent) @@ -1090,7 +1090,7 @@ func TestEncodeIntegerTypes(t *testing.T) { minValues.Rewind() maxValues.Rewind() codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1169,7 +1169,7 @@ func TestEncoderOtherTypes(t *testing.T) { } { event.Rewind() codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1246,7 +1246,7 @@ func TestE2EPartitionTableDMLBeforeDDL(t *testing.T) { common.EncodingFormatAvro, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) dec, err := NewDecoder(ctx, codecConfig, nil) @@ -1315,7 +1315,7 @@ func TestEncodeDMLBeforeDDL(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolSimple) - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) row, ok := event.GetNextRow() @@ -1399,7 +1399,7 @@ func TestEncodeBootstrapEvent(t *testing.T) { } { dmlEvent.Rewind() codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1476,7 +1476,7 @@ func TestEncodeLargeEventsNormal(t *testing.T) { } { codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, nil) @@ -1558,7 +1558,7 @@ func TestDDLMessageTooLarge(t *testing.T) { common.EncodingFormatJSON, } { codecConfig.EncodingFormat = format - enc, err := NewEncoder(context.Background(), codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) _, err = enc.EncodeDDLEvent(ddlEvent) @@ -1597,7 +1597,7 @@ func TestDMLMessageTooLarge(t *testing.T) { require.NoError(t, err) t.Cleanup(claimCheck.Close) } - enc, err := NewEncoder(context.Background(), codecConfig, claimCheck) + enc, err := NewEncoder(codecConfig, claimCheck) require.NoError(t, err) err = enc.AppendRowChangedEvent(context.Background(), "", insertEvent) @@ -1638,7 +1638,7 @@ func TestLargerMessageHandleClaimCheck(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, claimCheck) + enc, err := NewEncoder(codecConfig, claimCheck) require.NoError(t, err) m, err := enc.EncodeDDLEvent(ddlEvent) @@ -1718,7 +1718,7 @@ func TestLargeMessageHandleKeyOnly(t *testing.T) { codecConfig.MaxMessageBytes = config.DefaultMaxMessageBytes codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compressionType - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) rowEventDecoder, err := NewDecoder(ctx, codecConfig, db) @@ -1799,7 +1799,7 @@ func TestMarshallerError(t *testing.T) { ctx := context.Background() codecConfig := common.NewConfig(config.ProtocolSimple) - enc, err := NewEncoder(ctx, codecConfig, nil) + enc, err := NewEncoder(codecConfig, nil) require.NoError(t, err) mockMarshaller := mock_simple.NewMockmarshaller(gomock.NewController(t)) From 440252a36cdb6d7e2dea404a0b84fb9ccfbc7b2c Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Fri, 24 Jul 2026 18:03:15 +0800 Subject: [PATCH 8/8] simplify the code --- downstreamadapter/sink/kafka/helper.go | 69 ++++++++++++-------------- 1 file changed, 32 insertions(+), 37 deletions(-) diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go index 98c299bd4a..7d2e78ca54 100644 --- a/downstreamadapter/sink/kafka/helper.go +++ b/downstreamadapter/sink/kafka/helper.go @@ -60,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( @@ -99,52 +108,38 @@ func newKafkaSinkComponent( options.MaxMessageBytes, options.MaxBatchedBytes, ) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, 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) } - defer func() { - if err != nil { - claimCheck.Close() - } - }() - kafkaComponent.claimCheck = claimCheck - kafkaComponent.encoderGroup, err = codec.NewEncoderGroup(ctx, sinkConfig, encoderConfig, claimCheck, changefeedID) + 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.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck) + comp.encoder, err = codec.NewEventEncoder(ctx, encoderConfig, comp.claimCheck) if err != nil { - return kafkaComponent, protocol, errors.Trace(err) + return comp, protocol, errors.Trace(err) } - kafkaComponent.adminClient, err = kafkaComponent.factory.AdminClient(ctx) + comp.adminClient, err = comp.factory.AdminClient(ctx) if err != nil { - return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) + return comp, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err) } - // We must close adminClient when this func return cause by an error - // otherwise the adminClient will never be closed and lead to a goroutine leak. - defer func() { - if err != nil && kafkaComponent.adminClient != nil { - kafkaComponent.adminClient.Close() - } - }() - - kafkaComponent.topicManager, err = topicmanager.GetTopicManagerAndTryCreateTopic( + 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 }