kafka(ticdc): Revert "kafka(ticdc): claim check support large message…

… raw value for… (#11494)

close #11493

cdc/api/v2/model.go

@@ -352,7 +352,6 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 					LargeMessageHandleOption:      oldConfig.LargeMessageHandleOption,
 					LargeMessageHandleCompression: oldConfig.LargeMessageHandleCompression,
 					ClaimCheckStorageURI:          oldConfig.ClaimCheckStorageURI,
-					ClaimCheckRawValue:            oldConfig.ClaimCheckRawValue,
 				}
 			}
 
@@ -623,7 +622,6 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 					LargeMessageHandleOption:      oldConfig.LargeMessageHandleOption,
 					LargeMessageHandleCompression: oldConfig.LargeMessageHandleCompression,
 					ClaimCheckStorageURI:          oldConfig.ClaimCheckStorageURI,
-					ClaimCheckRawValue:            oldConfig.ClaimCheckRawValue,
 				}
 			}
 
@@ -991,7 +989,6 @@ type LargeMessageHandleConfig struct {
 	LargeMessageHandleOption      string `json:"large_message_handle_option"`
 	LargeMessageHandleCompression string `json:"large_message_handle_compression"`
 	ClaimCheckStorageURI          string `json:"claim_check_storage_uri"`
-	ClaimCheckRawValue            bool   `json:"claim_check_raw_value"`
 }
 
 // DispatchRule represents partition rule for a table

pkg/config/config_test_data.go

@@ -296,8 +296,7 @@ const (
       "large-message-handle": {
         "large-message-handle-option": "handle-key-only",
         "large-message-handle-compression": "",
-        "claim-check-storage-uri": "",
-		"claim-check-raw-value": false
+        "claim-check-storage-uri": ""
       },
       "glue-schema-registry-config": {
         "region":"region",

pkg/config/large_message.go

@@ -32,7 +32,6 @@ type LargeMessageHandleConfig struct {
 	LargeMessageHandleOption      string `toml:"large-message-handle-option" json:"large-message-handle-option"`
 	LargeMessageHandleCompression string `toml:"large-message-handle-compression" json:"large-message-handle-compression"`
 	ClaimCheckStorageURI          string `toml:"claim-check-storage-uri" json:"claim-check-storage-uri"`
-	ClaimCheckRawValue            bool   `toml:"claim-check-raw-value" json:"claim-check-raw-value"`
 }
 
 // NewDefaultLargeMessageHandleConfig return the default Config.
@@ -81,10 +80,6 @@ func (c *LargeMessageHandleConfig) AdjustAndValidate(protocol Protocol, enableTi
 			return cerror.ErrInvalidReplicaConfig.GenWithStack(
 				"large message handle is set to claim-check, but the claim-check-storage-uri is empty")
 		}
-		if c.ClaimCheckRawValue && protocol == ProtocolOpen {
-			return cerror.ErrInvalidReplicaConfig.GenWithStack(
-				"large message handle is set to claim-check, raw value is not supported for the open protocol")
-		}
 	}
 
 	return nil

pkg/config/large_message_test.go

@@ -58,7 +58,7 @@ func TestLargeMessageHandle4NotSupportedProtocol(t *testing.T) {
 	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
 }
 
-func TestHandleKeyOnly4CanalJSON(t *testing.T) {
+func TestLargeMessageHandle4CanalJSON(t *testing.T) {
 	t.Parallel()
 
 	// large-message-handle not set, always no error
@@ -68,45 +68,27 @@ func TestHandleKeyOnly4CanalJSON(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, largeMessageHandle.Disabled())
 
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionHandleKeyOnly
-
-	// `enable-tidb-extension` is false, return error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, false)
-	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
-
-	// `enable-tidb-extension` is true, no error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, true)
-	require.NoError(t, err)
-	require.Equal(t, LargeMessageHandleOptionHandleKeyOnly, largeMessageHandle.LargeMessageHandleOption)
-}
-
-func TestClaimCheck4CanalJSON(t *testing.T) {
-	t.Parallel()
+	for _, option := range []string{
+		LargeMessageHandleOptionHandleKeyOnly,
+		LargeMessageHandleOptionClaimCheck,
+	} {
+		largeMessageHandle.LargeMessageHandleOption = option
+		if option == LargeMessageHandleOptionClaimCheck {
+			largeMessageHandle.ClaimCheckStorageURI = "file:///tmp/claim-check"
+		}
 
-	// large-message-handle not set, always no error
-	largeMessageHandle := NewDefaultLargeMessageHandleConfig()
-
-	err := largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, false)
-	require.NoError(t, err)
-	require.True(t, largeMessageHandle.Disabled())
-
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionClaimCheck
-	largeMessageHandle.ClaimCheckStorageURI = "file:///tmp/claim-check"
-
-	for _, rawValue := range []bool{false, true} {
-		largeMessageHandle.ClaimCheckRawValue = rawValue
 		// `enable-tidb-extension` is false, return error
-		err = largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, false)
+		err := largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, false)
 		require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
 
 		// `enable-tidb-extension` is true, no error
 		err = largeMessageHandle.AdjustAndValidate(ProtocolCanalJSON, true)
 		require.NoError(t, err)
-		require.Equal(t, LargeMessageHandleOptionClaimCheck, largeMessageHandle.LargeMessageHandleOption)
+		require.Equal(t, option, largeMessageHandle.LargeMessageHandleOption)
 	}
 }
 
-func TestHandleKeyOnly4OpenProtocol(t *testing.T) {
+func TestLargeMessageHandle4OpenProtocol(t *testing.T) {
 	t.Parallel()
 
 	// large-message-handle not set, always no error
@@ -116,88 +98,23 @@ func TestHandleKeyOnly4OpenProtocol(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, largeMessageHandle.Disabled())
 
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionHandleKeyOnly
-	// `enable-tidb-extension` is false, return error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, false)
-	require.NoError(t, err)
-
-	// `enable-tidb-extension` is true, no error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, true)
-	require.NoError(t, err)
-	require.Equal(t, LargeMessageHandleOptionHandleKeyOnly, largeMessageHandle.LargeMessageHandleOption)
-}
-
-func TestClaimCheck4OpenProtocol(t *testing.T) {
-	t.Parallel()
-
-	// large-message-handle not set, always no error
-	largeMessageHandle := NewDefaultLargeMessageHandleConfig()
-
-	err := largeMessageHandle.AdjustAndValidate(ProtocolOpen, false)
-	require.NoError(t, err)
-	require.True(t, largeMessageHandle.Disabled())
-
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionClaimCheck
-	largeMessageHandle.ClaimCheckStorageURI = "file:///tmp/claim-check"
-
-	// `enable-tidb-extension` is false, return error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, false)
-	require.NoError(t, err)
-
-	// `enable-tidb-extension` is true, no error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, true)
-	require.NoError(t, err)
-	require.Equal(t, LargeMessageHandleOptionClaimCheck, largeMessageHandle.LargeMessageHandleOption)
-
-	largeMessageHandle.ClaimCheckRawValue = true
-	err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, true)
-	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
-}
-
-func TestHandleKeyOnly4SimpleProtocol(t *testing.T) {
-	t.Parallel()
-
-	// large-message-handle not set, always no error
-	largeMessageHandle := NewDefaultLargeMessageHandleConfig()
-
-	err := largeMessageHandle.AdjustAndValidate(ProtocolSimple, false)
-	require.NoError(t, err)
-	require.True(t, largeMessageHandle.Disabled())
-
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionHandleKeyOnly
-	// `enable-tidb-extension` is false, return error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolSimple, false)
-	require.NoError(t, err)
-
-	// `enable-tidb-extension` is true, no error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolSimple, true)
-	require.NoError(t, err)
-	require.Equal(t, LargeMessageHandleOptionHandleKeyOnly, largeMessageHandle.LargeMessageHandleOption)
-}
+	for _, o := range []string{
+		LargeMessageHandleOptionHandleKeyOnly,
+		LargeMessageHandleOptionClaimCheck,
+	} {
+		largeMessageHandle.LargeMessageHandleOption = o
+		if o == LargeMessageHandleOptionClaimCheck {
+			largeMessageHandle.ClaimCheckStorageURI = "file:///tmp/claim-check"
+		}
 
-func TestClaimCheck4SimpleProtocol(t *testing.T) {
-	t.Parallel()
-
-	// large-message-handle not set, always no error
-	largeMessageHandle := NewDefaultLargeMessageHandleConfig()
-
-	err := largeMessageHandle.AdjustAndValidate(ProtocolSimple, false)
-	require.NoError(t, err)
-	require.True(t, largeMessageHandle.Disabled())
-
-	largeMessageHandle.LargeMessageHandleOption = LargeMessageHandleOptionClaimCheck
-	largeMessageHandle.ClaimCheckStorageURI = "file:///tmp/claim-check"
-
-	// `enable-tidb-extension` is false, return error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolSimple, false)
-	require.NoError(t, err)
+		// `enable-tidb-extension` is false, return error
+		err := largeMessageHandle.AdjustAndValidate(ProtocolOpen, false)
+		require.NoError(t, err)
 
-	// `enable-tidb-extension` is true, no error
-	err = largeMessageHandle.AdjustAndValidate(ProtocolSimple, true)
-	require.NoError(t, err)
-	require.Equal(t, LargeMessageHandleOptionClaimCheck, largeMessageHandle.LargeMessageHandleOption)
+		// `enable-tidb-extension` is true, no error
+		err = largeMessageHandle.AdjustAndValidate(ProtocolOpen, true)
+		require.NoError(t, err)
+		require.Equal(t, o, largeMessageHandle.LargeMessageHandleOption)
 
-	largeMessageHandle.ClaimCheckRawValue = true
-	err = largeMessageHandle.AdjustAndValidate(ProtocolSimple, true)
-	require.NoError(t, err)
+	}
 }

pkg/sink/codec/canal/canal_json_decoder.go

@@ -143,16 +143,12 @@ func (b *batchDecoder) assembleClaimCheckRowChangedEvent(ctx context.Context, cl
 	if err != nil {
 		return nil, err
 	}
-
-	if !b.config.LargeMessageHandle.ClaimCheckRawValue {
-		claimCheckM, err := common.UnmarshalClaimCheckMessage(data)
-		if err != nil {
-			return nil, err
-		}
-		data = claimCheckM.Value
+	claimCheckM, err := common.UnmarshalClaimCheckMessage(data)
+	if err != nil {
+		return nil, err
 	}
 
-	value, err := common.Decompress(b.config.LargeMessageHandle.LargeMessageHandleCompression, data)
+	value, err := common.Decompress(b.config.LargeMessageHandle.LargeMessageHandleCompression, claimCheckM.Value)
 	if err != nil {
 		return nil, err
 	}

pkg/sink/codec/canal/canal_json_row_event_encoder.go

@@ -534,9 +534,15 @@ type jsonRowEventEncoderBuilder struct {
 
 // NewJSONRowEventEncoderBuilder creates a canal-json batchEncoderBuilder.
 func NewJSONRowEventEncoderBuilder(ctx context.Context, config *common.Config) (codec.RowEventEncoderBuilder, error) {
-	claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID)
-	if err != nil {
-		return nil, errors.Trace(err)
+	var (
+		claimCheck *claimcheck.ClaimCheck
+		err        error
+	)
+	if config.LargeMessageHandle.EnableClaimCheck() {
+		claimCheck, err = claimcheck.New(ctx, config.LargeMessageHandle.ClaimCheckStorageURI, config.ChangefeedID)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
 	}
 	return &jsonRowEventEncoderBuilder{
 		config:     config,

pkg/sink/codec/canal/canal_json_row_event_encoder_test.go

@@ -214,50 +214,46 @@ func TestCanalJSONClaimCheckE2E(t *testing.T) {
 	codecConfig.MaxMessageBytes = 500
 	ctx := context.Background()
 
-	for _, rawValue := range []bool{false, true} {
-		codecConfig.LargeMessageHandle.ClaimCheckRawValue = rawValue
-
-		builder, err := NewJSONRowEventEncoderBuilder(ctx, codecConfig)
-		require.NoError(t, err)
-		encoder := builder.Build()
+	builder, err := NewJSONRowEventEncoderBuilder(ctx, codecConfig)
+	require.NoError(t, err)
+	encoder := builder.Build()
 
-		_, insertEvent, _, _ := utils.NewLargeEvent4Test(t, config.GetDefaultReplicaConfig())
-		err = encoder.AppendRowChangedEvent(ctx, "", insertEvent, func() {})
-		require.NoError(t, err)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t, config.GetDefaultReplicaConfig())
+	err = encoder.AppendRowChangedEvent(ctx, "", insertEvent, func() {})
+	require.NoError(t, err)
 
-		// this is a large message, should be delivered to the external storage.
-		claimCheckLocationMessage := encoder.Build()[0]
+	// this is a large message, should be delivered to the external storage.
+	claimCheckLocationMessage := encoder.Build()[0]
 
-		decoder, err := NewBatchDecoder(ctx, codecConfig, nil)
-		require.NoError(t, err)
+	decoder, err := NewBatchDecoder(ctx, codecConfig, nil)
+	require.NoError(t, err)
 
-		err = decoder.AddKeyValue(claimCheckLocationMessage.Key, claimCheckLocationMessage.Value)
-		require.NoError(t, err)
+	err = decoder.AddKeyValue(claimCheckLocationMessage.Key, claimCheckLocationMessage.Value)
+	require.NoError(t, err)
 
-		messageType, ok, err := decoder.HasNext()
-		require.NoError(t, err)
-		require.Equal(t, messageType, model.MessageTypeRow)
-		require.True(t, ok)
+	messageType, ok, err := decoder.HasNext()
+	require.NoError(t, err)
+	require.Equal(t, messageType, model.MessageTypeRow)
+	require.True(t, ok)
 
-		decodedLargeEvent, err := decoder.NextRowChangedEvent()
-		require.NoError(t, err, rawValue)
+	decodedLargeEvent, err := decoder.NextRowChangedEvent()
+	require.NoError(t, err)
 
-		require.Equal(t, insertEvent.CommitTs, decodedLargeEvent.CommitTs)
-		require.Equal(t, insertEvent.TableInfo.GetSchemaName(), decodedLargeEvent.TableInfo.GetSchemaName())
-		require.Equal(t, insertEvent.TableInfo.GetTableName(), decodedLargeEvent.TableInfo.GetTableName())
-		require.Nil(t, nil, decodedLargeEvent.PreColumns)
+	require.Equal(t, insertEvent.CommitTs, decodedLargeEvent.CommitTs)
+	require.Equal(t, insertEvent.TableInfo.GetSchemaName(), decodedLargeEvent.TableInfo.GetSchemaName())
+	require.Equal(t, insertEvent.TableInfo.GetTableName(), decodedLargeEvent.TableInfo.GetTableName())
+	require.Nil(t, nil, decodedLargeEvent.PreColumns)
 
-		decodedColumns := make(map[string]*model.ColumnData, len(decodedLargeEvent.Columns))
-		for _, column := range decodedLargeEvent.Columns {
-			colName := decodedLargeEvent.TableInfo.ForceGetColumnName(column.ColumnID)
-			decodedColumns[colName] = column
-		}
-		for _, col := range insertEvent.Columns {
-			colName := insertEvent.TableInfo.ForceGetColumnName(col.ColumnID)
-			decoded, ok := decodedColumns[colName]
-			require.True(t, ok)
-			require.EqualValues(t, col.Value, decoded.Value)
-		}
+	decodedColumns := make(map[string]*model.ColumnData, len(decodedLargeEvent.Columns))
+	for _, column := range decodedLargeEvent.Columns {
+		colName := decodedLargeEvent.TableInfo.ForceGetColumnName(column.ColumnID)
+		decodedColumns[colName] = column
+	}
+	for _, col := range insertEvent.Columns {
+		colName := insertEvent.TableInfo.ForceGetColumnName(col.ColumnID)
+		decoded, ok := decodedColumns[colName]
+		require.True(t, ok)
+		require.EqualValues(t, col.Value, decoded.Value)
 	}
 }
 

pkg/sink/codec/open/open_protocol_encoder.go

@@ -351,9 +351,15 @@ func (b *batchEncoderBuilder) CleanMetrics() {
 func NewBatchEncoderBuilder(
 	ctx context.Context, config *common.Config,
 ) (codec.RowEventEncoderBuilder, error) {
-	claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID)
-	if err != nil {
-		return nil, errors.Trace(err)
+	var (
+		claimCheck *claimcheck.ClaimCheck
+		err        error
+	)
+	if config.LargeMessageHandle.EnableClaimCheck() {
+		claimCheck, err = claimcheck.New(ctx, config.LargeMessageHandle.ClaimCheckStorageURI, config.ChangefeedID)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
 	}
 	return &batchEncoderBuilder{
 		config:     config,

pkg/sink/codec/simple/decoder.go

@@ -170,16 +170,12 @@ func (d *Decoder) assembleClaimCheckRowChangedEvent(claimCheckLocation string) (
 	if err != nil {
 		return nil, err
 	}
-
-	if !d.config.LargeMessageHandle.ClaimCheckRawValue {
-		claimCheckM, err := common.UnmarshalClaimCheckMessage(data)
-		if err != nil {
-			return nil, err
-		}
-		data = claimCheckM.Value
+	claimCheckM, err := common.UnmarshalClaimCheckMessage(data)
+	if err != nil {
+		return nil, err
 	}
 
-	value, err := common.Decompress(d.config.LargeMessageHandle.LargeMessageHandleCompression, data)
+	value, err := common.Decompress(d.config.LargeMessageHandle.LargeMessageHandleCompression, claimCheckM.Value)
 	if err != nil {
 		return nil, err
 	}