kafka: decouple batch size from Kafka message size limit

downstreamadapter/sink/cloudstorage/encoder_group_test.go

@@ -230,6 +230,7 @@ func newTestTxnEncoderConfig(t *testing.T) *common.Config {
 		config.ProtocolCsv,
 		replicaConfig.Sink,
 		config.DefaultMaxMessageBytes,
+		config.DefaultMaxMessageBytes,
 	)
 	require.NoError(t, err)
 	return encoderConfig

downstreamadapter/sink/cloudstorage/sink.go

@@ -86,7 +86,7 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.
 	if err != nil {
 		return err
 	}
-	_, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt)
+	_, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt)
 	if err != nil {
 		return err
 	}
@@ -117,9 +117,9 @@ func New(
 	}
 	// get cloud storage file extension according to the specific protocol.
 	ext := helper.GetFileExtension(protocol)
-	// the last param maxMsgBytes is mainly to limit the size of a single message for
-	// batch protocols in mq scenario. In cloud storage sink, we just set it to max int.
-	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt)
+	// Message size limits are mainly for MQ batch protocols. Cloud storage uses
+	// max int for both the final message limit and the batch threshold.
+	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt)
 	if err != nil {
 		return nil, err
 	}

downstreamadapter/sink/helper/helper.go

@@ -50,17 +50,16 @@ func GetEncoderConfig(
 	sinkURI *url.URL,
 	protocol config.Protocol,
 	sinkConfig *config.SinkConfig,
-	maxMsgBytes int,
+	maxMessageBytes int,
+	maxBatchedBytes int,
 ) (*common.Config, error) {
 	encoderConfig := common.NewConfig(protocol)
 	if err := encoderConfig.Apply(sinkURI, sinkConfig); err != nil {
 		return nil, errors.WrapError(errors.ErrSinkInvalidConfig, err)
 	}
-	// Always set encoder's `MaxMessageBytes` equal to producer's `MaxMessageBytes`
-	// to prevent that the encoder generate batched message too large
-	// then cause producer meet `message too large`.
 	encoderConfig = encoderConfig.
-		WithMaxMessageBytes(maxMsgBytes).
+		WithMaxMessageBytes(maxMessageBytes).
+		WithMaxBatchedBytes(maxBatchedBytes).
 		WithChangefeedID(changefeedID)
 
 	tz, err := util.GetTimezone(config.GetGlobalServerConfig().TZ)

downstreamadapter/sink/kafka/helper.go

@@ -89,7 +89,10 @@ func newKafkaSinkComponent(
 		return kafkaComponent, protocol, errors.Trace(err)
 	}
 
-	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes)
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, sinkURI, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
 	if err != nil {
 		return kafkaComponent, protocol, errors.Trace(err)
 	}

downstreamadapter/sink/kafka/sink.go

@@ -87,7 +87,10 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.
 	}
 	options.Topic = topic
 
-	encoderConfig, err := helper.GetEncoderConfig(changefeedID, uri, protocol, sinkConfig, options.MaxMessageBytes)
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, uri, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
 	if err != nil {
 		return errors.Trace(err)
 	}

downstreamadapter/sink/kafka/sink_test.go

@@ -37,6 +37,19 @@ import (
 
 const kafkaSinkTestTopic = "mock_topic"
 
+func TestVerifyValidatesEncoderConfigBeforeKafkaConnection(t *testing.T) {
+	openProtocol := config.ProtocolOpen.String()
+	sinkConfig := &config.SinkConfig{Protocol: &openProtocol}
+	sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic + "?max-batch-size=0")
+	require.NoError(t, err)
+
+	changefeedID := common.NewChangefeedID4Test("test", "verify-existing-topic")
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	err = Verify(ctx, changefeedID, sinkURI, sinkConfig)
+	require.ErrorContains(t, err, "invalid max-batch-size 0")
+}
+
 func newKafkaSinkForTestWithProducers(ctx context.Context,
 	t *testing.T,
 	ctrl *gomock.Controller,
@@ -97,7 +110,10 @@ func newKafkaSinkForTestWithProducers(ctx context.Context,
 	if err != nil {
 		return nil, err
 	}
-	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes)
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, sinkURI, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
 	if err != nil {
 		return nil, err
 	}

downstreamadapter/sink/pulsar/helper.go

@@ -122,7 +122,10 @@ func newPulsarSinkComponentWithFactory(ctx context.Context,
 		return pulsarComponent, protocol, errors.Trace(err)
 	}
 
-	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, config.DefaultMaxMessageBytes)
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, sinkURI, protocol, sinkConfig,
+		config.DefaultMaxMessageBytes, config.DefaultMaxMessageBytes,
+	)
 	if err != nil {
 		return pulsarComponent, protocol, errors.Trace(err)
 	}

pkg/config/large_message.go

@@ -15,7 +15,7 @@ package config
 
 import (
 	"github.com/pingcap/ticdc/pkg/compression"
-	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/errors"
 )
 
 const (
@@ -55,34 +55,39 @@ func (c *LargeMessageHandleConfig) AdjustAndValidate(protocol Protocol, enableTi
 
 	// compression can be enabled independently
 	if !compression.Supported(c.LargeMessageHandleCompression) {
-		return cerror.ErrInvalidReplicaConfig.GenWithStack(
+		return errors.ErrInvalidReplicaConfig.GenWithStack(
 			"large message handle compression is not supported, got %s", c.LargeMessageHandleCompression)
 	}
 	if c.LargeMessageHandleOption == LargeMessageHandleOptionNone {
 		return nil
 	}
+	if c.LargeMessageHandleOption != LargeMessageHandleOptionClaimCheck &&
+		c.LargeMessageHandleOption != LargeMessageHandleOptionHandleKeyOnly {
+		return errors.ErrInvalidReplicaConfig.GenWithStack(
+			"unknown large-message-handle-option %s", c.LargeMessageHandleOption)
+	}
 
 	switch protocol {
 	case ProtocolOpen, ProtocolSimple:
 	case ProtocolCanalJSON:
 		if !enableTiDBExtension {
-			return cerror.ErrInvalidReplicaConfig.GenWithStack(
+			return errors.ErrInvalidReplicaConfig.GenWithStack(
 				"large message handle is set to %s, protocol is %s, but enable-tidb-extension is false",
 				c.LargeMessageHandleOption, protocol.String())
 		}
 	default:
-		return cerror.ErrInvalidReplicaConfig.GenWithStack(
+		return errors.ErrInvalidReplicaConfig.GenWithStack(
 			"large message handle is set to %s, protocol is %s, it's not supported",
 			c.LargeMessageHandleOption, protocol.String())
 	}
 
 	if c.LargeMessageHandleOption == LargeMessageHandleOptionClaimCheck {
 		if c.ClaimCheckStorageURI == "" {
-			return cerror.ErrInvalidReplicaConfig.GenWithStack(
+			return errors.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(
+			return errors.ErrInvalidReplicaConfig.GenWithStack(
 				"large message handle is set to claim-check, raw value is not supported for the open protocol")
 		}
 	}
@@ -106,7 +111,8 @@ func (c *LargeMessageHandleConfig) EnableClaimCheck() bool {
 	return c.LargeMessageHandleOption == LargeMessageHandleOptionClaimCheck
 }
 
-// Disabled returns true if disable large message handle.
+// Disabled returns true only when large message handling is explicitly disabled.
+// It returns false for nil and unknown configurations.
 func (c *LargeMessageHandleConfig) Disabled() bool {
 	if c == nil {
 		return false

pkg/config/large_message_test.go

@@ -58,6 +58,18 @@ func TestLargeMessageHandle4NotSupportedProtocol(t *testing.T) {
 	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
 }
 
+func TestLargeMessageHandleRejectsUnknownOption(t *testing.T) {
+	t.Parallel()
+
+	largeMessageHandle := NewDefaultLargeMessageHandleConfig()
+	largeMessageHandle.LargeMessageHandleOption = "unknown"
+
+	require.False(t, largeMessageHandle.Disabled())
+	err := largeMessageHandle.AdjustAndValidate(ProtocolOpen, false)
+	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
+	require.ErrorContains(t, err, "unknown large-message-handle-option unknown")
+}
+
 func TestHandleKeyOnly4CanalJSON(t *testing.T) {
 	t.Parallel()
 

pkg/sink/codec/canal/canal_json_txn_encoder.go

@@ -65,7 +65,6 @@ func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error
 			return err
 		}
 		length := len(value) + common.MaxRecordOverhead
-		// For single message that is longer than max-message-bytes, do not send it.
 		if length > j.config.MaxMessageBytes {
 			log.Warn("Single message is too large for canal-json",
 				zap.Int("maxMessageBytes", j.config.MaxMessageBytes),

pkg/sink/codec/common/config.go

@@ -41,9 +41,12 @@ type Config struct {
 
 	Protocol config.Protocol
 
-	// control batch behavior, only for `open-protocol` and `craft` at the moment.
 	MaxMessageBytes int
-	MaxBatchSize    int
+
+	// MaxBatchedBytes controls open-protocol encoder's maximum number of bytes for a batched message.
+	MaxBatchedBytes int
+	// MaxBatchedBytes controls open-protocol encoder's maximum number of events for a batched message.
+	MaxBatchSize int
 
 	// DeleteOnlyHandleKeyColumns is true, for the delete event only output the handle key columns.
 	DeleteOnlyHandleKeyColumns bool
@@ -115,6 +118,7 @@ func NewConfig(protocol config.Protocol) *Config {
 		Protocol: protocol,
 
 		MaxMessageBytes: config.DefaultMaxMessageBytes,
+		MaxBatchedBytes: config.DefaultMaxMessageBytes,
 		MaxBatchSize:    defaultMaxBatchSize,
 
 		EnableTiDBExtension: false,
@@ -194,7 +198,7 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error {
 	var err error
 	urlParameter := &urlConfig{}
 	if err = binding.Query.Bind(req, urlParameter); err != nil {
-		return errors.WrapError(errors.ErrMySQLInvalidConfig, err)
+		return errors.WrapError(errors.ErrSinkInvalidConfig, err)
 	}
 	if urlParameter, err = mergeConfig(sinkConfig, urlParameter); err != nil {
 		return err
@@ -347,6 +351,12 @@ func (c *Config) WithMaxMessageBytes(bytes int) *Config {
 	return c
 }
 
+// WithMaxBatchedBytes sets the maximum batched message bytes.
+func (c *Config) WithMaxBatchedBytes(bytes int) *Config {
+	c.MaxBatchedBytes = bytes
+	return c
+}
+
 // WithChangefeedID set the `changefeedID`
 func (c *Config) WithChangefeedID(id common.ChangeFeedID) *Config {
 	c.ChangefeedID = id
@@ -457,15 +467,17 @@ func (c *Config) Validate() error {
 	}
 
 	if c.MaxMessageBytes <= 0 {
-		return errors.ErrCodecInvalidConfig.Wrap(
-			errors.Errorf("invalid max-message-bytes %d", c.MaxMessageBytes),
-		)
+		return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-message-bytes %d", c.MaxMessageBytes)
+	}
+	if c.MaxBatchedBytes < 0 {
+		return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-batch-message-bytes %d", c.MaxBatchedBytes)
+	}
+	if c.MaxBatchedBytes > c.MaxMessageBytes {
+		return errors.ErrCodecInvalidConfig.GenWithStack("max-batch-message-bytes %d cannot be greater than max-message-bytes %d", c.MaxBatchedBytes, c.MaxMessageBytes)
 	}
 
 	if c.MaxBatchSize <= 0 {
-		return errors.ErrCodecInvalidConfig.Wrap(
-			errors.Errorf("invalid max-batch-size %d", c.MaxBatchSize),
-		)
+		return errors.ErrCodecInvalidConfig.GenWithStack("invalid max-batch-size %d", c.MaxBatchSize)
 	}
 
 	if c.LargeMessageHandle != nil {

pkg/sink/codec/common/config_test.go

@@ -18,10 +18,73 @@ import (
 	"testing"
 
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/stretchr/testify/require"
 )
 
+func TestApplyReturnsSinkInvalidConfigForQueryBindingError(t *testing.T) {
+	cfg := NewConfig(config.ProtocolOpen)
+	sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?max-batch-size=invalid")
+	require.NoError(t, err)
+
+	err = cfg.Apply(sinkURI, config.GetDefaultReplicaConfig().Sink)
+	errCode, ok := errors.RFCCode(err)
+	require.True(t, ok, err)
+	require.Equal(t, errors.ErrSinkInvalidConfig.RFCCode(), errCode)
+}
+
+func TestValidateMaxBatchMessageBytes(t *testing.T) {
+	tests := []struct {
+		name     string
+		adjust   func(*Config)
+		expected string
+	}{
+		{
+			name: "non-positive max message bytes",
+			adjust: func(cfg *Config) {
+				cfg.MaxMessageBytes = 0
+			},
+			expected: "invalid max-message-bytes 0",
+		},
+		{
+			name: "negative max batched bytes",
+			adjust: func(cfg *Config) {
+				cfg.MaxBatchedBytes = -1
+			},
+			expected: "invalid max-batch-message-bytes -1",
+		},
+		{
+			name: "max batched bytes exceeds max message bytes",
+			adjust: func(cfg *Config) {
+				cfg.MaxMessageBytes = 100
+				cfg.MaxBatchedBytes = 101
+			},
+			expected: "max-batch-message-bytes 101 cannot be greater than max-message-bytes 100",
+		},
+		{
+			name: "non-positive max batch size",
+			adjust: func(cfg *Config) {
+				cfg.MaxBatchSize = 0
+			},
+			expected: "invalid max-batch-size 0",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := NewConfig(config.ProtocolOpen)
+			test.adjust(cfg)
+
+			err := cfg.Validate()
+			require.ErrorContains(t, err, test.expected)
+			errCode, ok := errors.RFCCode(err)
+			require.True(t, ok, err)
+			require.Equal(t, errors.ErrCodecInvalidConfig.RFCCode(), errCode)
+		})
+	}
+}
+
 func TestDebeziumAvroSchemaRegistryConfig(t *testing.T) {
 	t.Parallel()
 

pkg/sink/codec/open/encoder.go

@@ -38,7 +38,7 @@ var (
 )
 
 // batchEncoder for open protocol will batch multiple row changed events into a single message.
-// One message can contain at most MaxBatchSize events, and the total size of the message cannot exceed MaxMessageBytes.
+// One message can contain at most MaxBatchSize events, and the total size of the message cannot exceed MaxBatchedBytes.
 type batchEncoder struct {
 	messages []*common.Message
 	// buff the callback of the latest message
@@ -174,7 +174,7 @@ func (d *batchEncoder) pushMessage(key, value []byte, callback func()) {
 	binary.BigEndian.PutUint64(keyLenByte[:], uint64(len(key)))
 	binary.BigEndian.PutUint64(valueLenByte[:], uint64(len(value)))
 
-	if len(d.messages) == 0 || d.messages[len(d.messages)-1].Length()+length > d.config.MaxMessageBytes || d.messages[len(d.messages)-1].GetRowsCount() >= d.config.MaxBatchSize {
+	if len(d.messages) == 0 || d.messages[len(d.messages)-1].Length()+length > d.config.MaxBatchedBytes || d.messages[len(d.messages)-1].GetRowsCount() >= d.config.MaxBatchSize {
 		d.finalizeCallback()
 		// create a new message
 		versionHead := make([]byte, 8)