diff --git a/downstreamadapter/sink/cloudstorage/encoder_group_test.go b/downstreamadapter/sink/cloudstorage/encoder_group_test.go
index 269702ec04..ccc722df03 100644
--- a/downstreamadapter/sink/cloudstorage/encoder_group_test.go
+++ b/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
diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go
index eb5519772a..79204a8901 100644
--- a/downstreamadapter/sink/cloudstorage/sink.go
+++ b/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
 	}
diff --git a/downstreamadapter/sink/helper/helper.go b/downstreamadapter/sink/helper/helper.go
index 11dd7b85de..f93410f9ab 100644
--- a/downstreamadapter/sink/helper/helper.go
+++ b/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)
diff --git a/downstreamadapter/sink/kafka/helper.go b/downstreamadapter/sink/kafka/helper.go
index 6665b4c973..bd776c5e49 100644
--- a/downstreamadapter/sink/kafka/helper.go
+++ b/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)
 	}
diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go
index 673d3e9993..9de1070e15 100644
--- a/downstreamadapter/sink/kafka/sink.go
+++ b/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)
 	}
diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go
index 440806501c..6d7d337574 100644
--- a/downstreamadapter/sink/kafka/sink_test.go
+++ b/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
 	}
diff --git a/downstreamadapter/sink/pulsar/helper.go b/downstreamadapter/sink/pulsar/helper.go
index d2c33430b5..3ccb299767 100644
--- a/downstreamadapter/sink/pulsar/helper.go
+++ b/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)
 	}
diff --git a/pkg/config/large_message.go b/pkg/config/large_message.go
index d04584b451..6b19afe260 100644
--- a/pkg/config/large_message.go
+++ b/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
diff --git a/pkg/config/large_message_test.go b/pkg/config/large_message_test.go
index ad3345643e..f0721f4b53 100644
--- a/pkg/config/large_message_test.go
+++ b/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()
 
diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder.go b/pkg/sink/codec/canal/canal_json_txn_encoder.go
index 0af6f4f3f2..3e10d98229 100644
--- a/pkg/sink/codec/canal/canal_json_txn_encoder.go
+++ b/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),
diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go
index 2a41e5b2fd..4f5cee429d 100644
--- a/pkg/sink/codec/common/config.go
+++ b/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 {
diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go
index c138b4ddf4..4369de7216 100644
--- a/pkg/sink/codec/common/config_test.go
+++ b/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()
 
diff --git a/pkg/sink/codec/open/encoder.go b/pkg/sink/codec/open/encoder.go
index 75f82e9d03..17a323cb7d 100644
--- a/pkg/sink/codec/open/encoder.go
+++ b/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)
diff --git a/pkg/sink/codec/open/encoder_test.go b/pkg/sink/codec/open/encoder_test.go
index 9b02366709..55f3f37c86 100644
--- a/pkg/sink/codec/open/encoder_test.go
+++ b/pkg/sink/codec/open/encoder_test.go
@@ -189,7 +189,7 @@ func TestFloatTypes(t *testing.T) {
 	helper.Tk().MustExec("use test")
 	job := helper.DDL2Job(`create table test.t(
     	id int primary key auto_increment,
-	    a float, b float(10, 3), c float(10), 
+	    a float, b float(10, 3), c float(10),
 	    d double, e double(20, 3))`)
 
 	dmlEvent := helper.DML2Event("test", "t", `insert into test.t(a,b,c,d,e) values (1.23, 4.56, 7.89, 10.11, 12.13)`)
@@ -533,17 +533,17 @@ func TestOtherTypes(t *testing.T) {
 
 	helper.Tk().MustExec("use test")
 	job := helper.DDL2Job(`create table test.t(
-    	id int primary key auto_increment, 
+		id int primary key auto_increment,
     	a bool, b bool, c year,
-		d bit(10), e json, 
-		f decimal(10,2), 
+		d bit(10), e json,
+		f decimal(10,2),
 		g enum('a','b','c'), h set('a','b','c'))`)
 	tableInfo := helper.GetTableInfo(job)
 
 	dmlEvent := helper.DML2Event("test", "t", `insert into test.t(a, b, c, d, e, f, g, h) values (
-   		true, false, 2000, 
-	    0b0101010101, '{"key1": "value1"}', 
-	    153.123, 
+		true, false, 2000,
+	    0b0101010101, '{"key1": "value1"}',
+	    153.123,
 	    'a', 'a,b')`)
 
 	require.NotNil(t, dmlEvent)
@@ -778,7 +778,9 @@ func TestEncoderMultipleMessage(t *testing.T) {
 		`insert into test.t values (3, 333)`)
 
 	ctx := context.Background()
-	codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(400)
+	codecConfig := common.NewConfig(config.ProtocolOpen).
+		WithMaxMessageBytes(1000).
+		WithMaxBatchedBytes(400)
 	encoder, err := NewBatchEncoder(ctx, codecConfig)
 	require.NoError(t, err)
 
@@ -808,11 +810,13 @@ func TestEncoderMultipleMessage(t *testing.T) {
 	require.Equal(t, 2, len(messages))
 	require.Equal(t, 2, messages[0].GetRowsCount())
 	require.Equal(t, 1, messages[1].GetRowsCount())
+	require.LessOrEqual(t, messages[0].Length(), codecConfig.MaxBatchedBytes)
+	require.LessOrEqual(t, messages[1].Length(), codecConfig.MaxBatchedBytes)
+	require.Equal(t, 0, count)
 
-	for _, message := range messages {
-		message.Callback()
-	}
-
+	messages[0].Callback()
+	require.Equal(t, 2, count)
+	messages[1].Callback()
 	require.Equal(t, 3, count)
 
 	decoder, err := NewDecoder(ctx, 0, codecConfig, nil)
@@ -885,6 +889,48 @@ func TestMessageTooLarge(t *testing.T) {
 	require.Equal(t, count, 0)
 }
 
+func TestMessageLargerThanBatchLimit(t *testing.T) {
+	ctx := context.Background()
+	codecConfig := common.NewConfig(config.ProtocolOpen).
+		WithMaxMessageBytes(400).
+		WithMaxBatchedBytes(100)
+	encoder, err := NewBatchEncoder(ctx, codecConfig)
+	require.NoError(t, err)
+
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+	helper.Tk().MustExec("use test")
+
+	job := helper.DDL2Job(`create table test.t(a tinyint primary key, b int)`)
+	tableInfo := helper.GetTableInfo(job)
+	dmlEvent := helper.DML2Event("test", "t", `insert into test.t values (1, 123)`)
+	require.NotNil(t, dmlEvent)
+	insertRow, ok := dmlEvent.GetNextRow()
+	require.True(t, ok)
+
+	count := 0
+	insertRowEvent := &commonEvent.RowEvent{
+		TableInfo:      tableInfo,
+		CommitTs:       dmlEvent.GetCommitTs(),
+		Event:          insertRow,
+		ColumnSelector: columnselector.NewDefaultColumnSelector(),
+		Callback:       func() { count += 1 },
+	}
+
+	err = encoder.AppendRowChangedEvent(ctx, "", insertRowEvent)
+	require.NoError(t, err)
+
+	messages := encoder.Build()
+	require.Len(t, messages, 1)
+	require.Equal(t, 1, messages[0].GetRowsCount())
+	require.Greater(t, messages[0].Length(), codecConfig.MaxBatchedBytes)
+	require.LessOrEqual(t, messages[0].Length(), codecConfig.MaxMessageBytes)
+	require.Equal(t, 0, count)
+
+	messages[0].Callback()
+	require.Equal(t, 1, count)
+}
+
 func TestLargeMessageWithHandleEnableHandleKeyOnly(t *testing.T) {
 	helper := commonEvent.NewEventTestHelper(t)
 	defer helper.Close()
diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go
index 7c51166a9e..301471917c 100644
--- a/pkg/sink/kafka/options.go
+++ b/pkg/sink/kafka/options.go
@@ -26,11 +26,10 @@ import (
 
 	"github.com/gin-gonic/gin/binding"
 	"github.com/imdario/mergo"
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
-	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/security"
 	"go.uber.org/zap"
 )
@@ -40,13 +39,6 @@ const (
 	defaultPartitionNum = 3
 	// defaultMaxRetry is the default retry budget for Kafka producers.
 	defaultMaxRetry = 5
-
-	// the `max-message-bytes` is set equal to topic's `max.message.bytes`, and is used to check
-	// whether the message is larger than the max size limit. It's found some message pass the message
-	// size limit check at the client side and failed at the broker side since message enlarged during
-	// the network transmission. so we set the `max-message-bytes` to a smaller value to avoid this problem.
-	// maxMessageBytesOverhead is used to reduce the `max-message-bytes`.
-	maxMessageBytesOverhead = 128
 )
 
 const (
@@ -108,7 +100,7 @@ func requireAcksFromString(acks int) (RequiredAcks, error) {
 	case int(NoResponse):
 		return NoResponse, nil
 	default:
-		return Unknown, cerror.ErrKafkaInvalidRequiredAcks.GenWithStackByArgs(acks)
+		return Unknown, errors.ErrKafkaInvalidRequiredAcks.GenWithStackByArgs(acks)
 	}
 }
 
@@ -143,7 +135,7 @@ type urlConfig struct {
 	InsecureSkipVerify           *bool   `form:"insecure-skip-verify"`
 }
 
-// options stores user specified configurations
+// options stores Kafka sink configurations
 type options struct {
 	Topic           string
 	BrokerEndpoints []string
@@ -156,14 +148,16 @@ type options struct {
 	Version           string
 	IsAssignedVersion bool
 	RequestVersion    int16
-	MaxMessageBytes   int
-	MaxRetry          int
-	Compression       string
-	ClientID          string
-	RequiredAcks      RequiredAcks
-	// Only for test. User can not set this value.
-	// The current prod default value is 0.
-	MaxMessages int
+
+	// MaxMessageBytes controls the byte size limit of the producer.
+	MaxMessageBytes int
+	// MaxBatchedBytes controls the byte size limit when batching messages.
+	MaxBatchedBytes int
+
+	MaxRetry     int
+	Compression  string
+	ClientID     string
+	RequiredAcks RequiredAcks
 
 	// Credential is used to connect to kafka cluster.
 	EnableTLS          bool
@@ -180,9 +174,9 @@ type options struct {
 // NewOptions returns a default Kafka configuration
 func NewOptions() *options {
 	return &options{
-		Version: "2.4.0",
-		// MaxMessageBytes will be used to initialize producer
+		Version:            "2.4.0",
 		MaxMessageBytes:    config.DefaultMaxMessageBytes,
+		MaxBatchedBytes:    config.DefaultMaxMessageBytes,
 		MaxRetry:           defaultMaxRetry,
 		ReplicationFactor:  1,
 		Compression:        "none",
@@ -198,11 +192,12 @@ func NewOptions() *options {
 }
 
 // setPartitionNum set the partition-num by the topic's partition count.
-func (o *options) setPartitionNum(realPartitionCount int32) error {
+func (o *options) setPartitionNum(changefeedID common.ChangeFeedID, realPartitionCount int32) error {
 	// user does not specify the `partition-num` in the sink-uri
 	if o.PartitionNum == 0 {
 		o.PartitionNum = realPartitionCount
 		log.Info("partitionNum is not set, set by topic's partition-num",
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
 			zap.Int32("partitionNum", realPartitionCount))
 		return nil
 	}
@@ -210,8 +205,8 @@ func (o *options) setPartitionNum(realPartitionCount int32) error {
 	if o.PartitionNum < realPartitionCount {
 		log.Warn("number of partition specified in sink-uri is less than that of the actual topic. "+
 			"Some partitions will not have messages dispatched to",
-			zap.Int32("sinkUriPartitions", o.PartitionNum),
-			zap.Int32("topicPartitions", realPartitionCount))
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
+			zap.Int32("sinkUriPartitions", o.PartitionNum), zap.Int32("topicPartitions", realPartitionCount))
 		return nil
 	}
 
@@ -219,7 +214,7 @@ func (o *options) setPartitionNum(realPartitionCount int32) error {
 	// the real partition count, since messages would be dispatched to different
 	// partitions, this could prevent potential correctness problems.
 	if o.PartitionNum > realPartitionCount {
-		return cerror.ErrKafkaInvalidPartitionNum.GenWithStack(
+		return errors.ErrKafkaInvalidPartitionNum.GenWithStack(
 			"the number of partition (%d) specified in sink-uri is more than that of actual topic (%d)",
 			o.PartitionNum, realPartitionCount)
 	}
@@ -236,7 +231,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID,
 	req := &http.Request{URL: sinkURI}
 	urlParameter := &urlConfig{}
 	if err = binding.Query.Bind(req, urlParameter); err != nil {
-		return cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
+		return errors.WrapError(errors.ErrMySQLInvalidConfig, err)
 	}
 	if urlParameter, err = mergeConfig(sinkConfig, urlParameter); err != nil {
 		return err
@@ -244,7 +239,7 @@ func (o *options) Apply(changefeedID common.ChangeFeedID,
 	if urlParameter.PartitionNum != nil {
 		o.PartitionNum = *urlParameter.PartitionNum
 		if o.PartitionNum <= 0 {
-			return cerror.ErrKafkaInvalidPartitionNum.GenWithStackByArgs(o.PartitionNum)
+			return errors.ErrKafkaInvalidPartitionNum.GenWithStackByArgs(o.PartitionNum)
 		}
 	}
 
@@ -258,8 +253,12 @@ func (o *options) Apply(changefeedID common.ChangeFeedID,
 	}
 
 	if urlParameter.MaxMessageBytes != nil {
+		if *urlParameter.MaxMessageBytes <= 0 {
+			return errors.ErrKafkaInvalidConfig.GenWithStack("invalid max-message-bytes %d", *urlParameter.MaxMessageBytes)
+		}
 		o.MaxMessageBytes = *urlParameter.MaxMessageBytes
 	}
+	o.MaxBatchedBytes = o.MaxMessageBytes
 
 	if urlParameter.MaxRetry != nil && *urlParameter.MaxRetry >= 0 {
 		o.MaxRetry = *urlParameter.MaxRetry
@@ -387,7 +386,7 @@ func (o *options) applyTLS(params *urlConfig) error {
 
 	if o.Credential != nil && !o.Credential.IsEmpty() &&
 		!o.Credential.IsTLSEnabled() {
-		return cerror.WrapError(cerror.ErrKafkaInvalidConfig,
+		return errors.WrapError(errors.ErrKafkaInvalidConfig,
 			errors.New("ca, cert and key files should all be supplied"))
 	}
 
@@ -401,7 +400,7 @@ func (o *options) applyTLS(params *urlConfig) error {
 		enableTLS := *params.EnableTLS
 
 		if o.Credential != nil && o.Credential.IsTLSEnabled() && !enableTLS {
-			return cerror.WrapError(cerror.ErrKafkaInvalidConfig,
+			return errors.WrapError(errors.ErrKafkaInvalidConfig,
 				errors.New("credential files are supplied, but 'enable-tls' is set to false"))
 		}
 		o.EnableTLS = enableTLS
@@ -431,7 +430,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 	if urlParameter.SASLMechanism != nil && *urlParameter.SASLMechanism != "" {
 		mechanism, err := security.SASLMechanismFromString(*urlParameter.SASLMechanism)
 		if err != nil {
-			return cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+			return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
 		}
 		o.SASL.SASLMechanism = mechanism
 	}
@@ -439,7 +438,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 	if urlParameter.SASLGssAPIAuthType != nil && *urlParameter.SASLGssAPIAuthType != "" {
 		authType, err := security.AuthTypeFromString(*urlParameter.SASLGssAPIAuthType)
 		if err != nil {
-			return cerror.WrapError(cerror.ErrKafkaInvalidConfig, err)
+			return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
 		}
 		o.SASL.GSSAPI.AuthType = authType
 	}
@@ -477,7 +476,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 		if sinkConfig.KafkaConfig.SASLOAuthClientID != nil {
 			clientID := *sinkConfig.KafkaConfig.SASLOAuthClientID
 			if clientID == "" {
-				return cerror.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client ID cannot be empty")
+				return errors.ErrKafkaInvalidConfig.GenWithStack("OAuth2 client ID cannot be empty")
 			}
 			o.SASL.OAuth2.ClientID = clientID
 		}
@@ -485,7 +484,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 		if sinkConfig.KafkaConfig.SASLOAuthClientSecret != nil {
 			clientSecret := *sinkConfig.KafkaConfig.SASLOAuthClientSecret
 			if clientSecret == "" {
-				return cerror.ErrKafkaInvalidConfig.GenWithStack(
+				return errors.ErrKafkaInvalidConfig.GenWithStack(
 					"OAuth2 client secret cannot be empty")
 			}
 
@@ -493,7 +492,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 			decodedClientSecret, err := base64.StdEncoding.DecodeString(clientSecret)
 			if err != nil {
 				log.Error("OAuth2 client secret is not base64 encoded", zap.Error(err))
-				return cerror.ErrKafkaInvalidConfig.GenWithStack(
+				return errors.ErrKafkaInvalidConfig.GenWithStack(
 					"OAuth2 client secret is not base64 encoded")
 			}
 			o.SASL.OAuth2.ClientSecret = string(decodedClientSecret)
@@ -502,7 +501,7 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 		if sinkConfig.KafkaConfig.SASLOAuthTokenURL != nil {
 			tokenURL := *sinkConfig.KafkaConfig.SASLOAuthTokenURL
 			if tokenURL == "" {
-				return cerror.ErrKafkaInvalidConfig.GenWithStack(
+				return errors.ErrKafkaInvalidConfig.GenWithStack(
 					"OAuth2 token URL cannot be empty")
 			}
 			o.SASL.OAuth2.TokenURL = tokenURL
@@ -510,13 +509,13 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 
 		if o.SASL.OAuth2.IsEnable() {
 			if o.SASL.SASLMechanism != security.OAuthMechanism {
-				return cerror.ErrKafkaInvalidConfig.GenWithStack(
+				return errors.ErrKafkaInvalidConfig.GenWithStack(
 					"OAuth2 is only supported with SASL mechanism type OAUTHBEARER, but got %s",
 					o.SASL.SASLMechanism)
 			}
 
 			if err := o.SASL.OAuth2.Validate(); err != nil {
-				return cerror.ErrKafkaInvalidConfig.Wrap(err)
+				return errors.ErrKafkaInvalidConfig.Wrap(err)
 			}
 			o.SASL.OAuth2.SetDefault()
 		}
@@ -570,14 +569,17 @@ func NewKafkaClientID(captureAddr string,
 		clientID = commonInvalidChar.ReplaceAllString(clientID, "_")
 	}
 	if !validClientID.MatchString(clientID) {
-		return "", cerror.ErrKafkaInvalidClientID.GenWithStackByArgs(clientID)
+		return "", errors.ErrKafkaInvalidClientID.GenWithStackByArgs(clientID)
 	}
 	return
 }
 
-// adjustOptions adjust the `options` and `sarama.Config` by condition.
+// adjustOptions adjusts options with Kafka runtime metadata.
+// It overwrites MaxMessageBytes with the final producer message limit derived
+// from the topic or broker configuration.
 func adjustOptions(
 	ctx context.Context,
+	changefeedID common.ChangeFeedID,
 	admin ClusterAdminClient,
 	options *options,
 	topic string,
@@ -587,102 +589,140 @@ func adjustOptions(
 		return errors.Trace(err)
 	}
 
-	// Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1.
-	// If we don't check it, the producer probably can not send message to the topic.
-	// Because it will wait for the ack from all replicas. But we do not have enough replicas.
-	if options.RequiredAcks == WaitForAll {
-		err = validateMinInsyncReplicas(ctx, admin, topics, topic, int(options.ReplicationFactor))
-		if err != nil {
-			return errors.Trace(err)
-		}
+	if err = validateRequiredAcks(ctx, admin, topics, topic, options); err != nil {
+		return errors.Trace(err)
 	}
+	return adjustTopicOptions(ctx, changefeedID, admin, options, topic, topics)
+}
 
+func adjustTopicOptions(
+	ctx context.Context,
+	changefeedID common.ChangeFeedID,
+	admin ClusterAdminClient,
+	options *options,
+	topic string,
+	topics map[string]TopicDetail,
+) error {
 	info, exists := topics[topic]
 	// once we have found the topic, no matter `auto-create-topic`,
 	// make sure user input parameters are valid.
+	var err error
 	if exists {
-		// make sure that producer's `MaxMessageBytes` smaller than topic's `max.message.bytes`
-		topicMaxMessageBytesStr, err := getTopicConfig(
-			ctx, admin, info.Name,
-			TopicMaxMessageBytesConfigName,
-			BrokerMessageMaxBytesConfigName,
-		)
-		var topicMaxMessageBytes int
-		if err != nil {
-			log.Warn("TiCDC cannot find `max.message.bytes` from topic's configuration, use the option `MaxMessageBytes` as default")
-			topicMaxMessageBytes = options.MaxMessageBytes
-		} else {
-			topicMaxMessageBytes, err = strconv.Atoi(topicMaxMessageBytesStr)
-			if err != nil {
-				return errors.Trace(err)
-			}
-		}
-
-		maxMessageBytes := topicMaxMessageBytes - maxMessageBytesOverhead
-		if topicMaxMessageBytes <= options.MaxMessageBytes {
-			log.Warn("topic's `max.message.bytes` less than the `max-message-bytes`,"+
-				"use topic's `max.message.bytes` to initialize the Kafka producer",
-				zap.Int("max.message.bytes", topicMaxMessageBytes),
-				zap.Int("max-message-bytes", options.MaxMessageBytes),
-				zap.Int("real-max-message-bytes", maxMessageBytes))
-			options.MaxMessageBytes = maxMessageBytes
-		} else {
-			if maxMessageBytes < options.MaxMessageBytes {
-				options.MaxMessageBytes = maxMessageBytes
-			}
-		}
-
-		// no need to create the topic,
-		// but we would have to log user if they found enter wrong topic name later
-		if options.AutoCreate {
-			log.Warn("topic already exist, TiCDC will not create the topic",
-				zap.String("topic", topic), zap.Any("detail", info))
-		}
+		err = adjustExistingTopicOption(ctx, changefeedID, admin, options, topic, info)
+	} else {
+		adjustNewTopicOptions(admin, changefeedID, options, topic)
+	}
+	if err != nil {
+		return err
+	}
 
-		if err = options.setPartitionNum(info.NumPartitions); err != nil {
-			return errors.Trace(err)
-		}
+	options.MaxBatchedBytes = min(options.MaxBatchedBytes, options.MaxMessageBytes)
+	return nil
+}
 
+func validateRequiredAcks(
+	ctx context.Context,
+	admin ClusterAdminClient,
+	topics map[string]TopicDetail,
+	topic string,
+	options *options,
+) error {
+	// Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1.
+	// If we don't check it, the producer probably can not send message to the topic.
+	// Because it will wait for the ack from all replicas. But we do not have enough replicas.
+	if options.RequiredAcks != WaitForAll {
 		return nil
 	}
+	return validateMinInsyncReplicas(ctx, admin, topics, topic, int(options.ReplicationFactor))
+}
 
-	var brokerMessageMaxBytes int
-	brokerMessageMaxBytesStr, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName)
+func adjustExistingTopicOption(
+	ctx context.Context,
+	changefeedID common.ChangeFeedID,
+	admin ClusterAdminClient,
+	options *options,
+	topic string,
+	info TopicDetail,
+) error {
+	maxMessageBytes, err := getTopicMaxMessageBytes(ctx, admin, info.Name)
 	if err != nil {
-		log.Warn("TiCDC cannot find `message.max.bytes` from broker's configuration, use the option `MaxMessageBytes` as default")
-		brokerMessageMaxBytes = options.MaxMessageBytes
-	} else {
-		brokerMessageMaxBytes, err = strconv.Atoi(brokerMessageMaxBytesStr)
-		if err != nil {
-			return errors.Trace(err)
-		}
+		log.Warn("`max.message.bytes` not found from topic's configuration, use the option `MaxMessageBytes` as default",
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
+			zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err))
+		maxMessageBytes = options.MaxMessageBytes
+	}
+	options.MaxMessageBytes = maxMessageBytes
+
+	// no need to create the topic,
+	// but we would have to log user if they found enter wrong topic name later
+	if options.AutoCreate {
+		log.Warn("topic already exist, TiCDC will not create the topic",
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
+			zap.String("topic", topic), zap.Any("detail", info))
 	}
 
+	if err = options.setPartitionNum(changefeedID, info.NumPartitions); err != nil {
+		return errors.Trace(err)
+	}
+	return nil
+}
+
+func adjustNewTopicOptions(
+	admin ClusterAdminClient,
+	changefeedID common.ChangeFeedID,
+	options *options,
+	topic string,
+) {
 	// when create the topic, `max.message.bytes` is decided by the broker,
 	// it would use broker's `message.max.bytes` to set topic's `max.message.bytes`.
-	// TiCDC need to make sure that the producer's `MaxMessageBytes` won't larger than
-	// broker's `message.max.bytes`.
-	maxMessageBytes := brokerMessageMaxBytes - maxMessageBytesOverhead
-	if brokerMessageMaxBytes <= options.MaxMessageBytes {
-		log.Warn("broker's `message.max.bytes` less than the `max-message-bytes`,"+
-			"use broker's `message.max.bytes` to initialize the Kafka producer",
-			zap.Int("message.max.bytes", brokerMessageMaxBytes),
-			zap.Int("max-message-bytes", options.MaxMessageBytes),
-			zap.Int("real-max-message-bytes", maxMessageBytes))
-		options.MaxMessageBytes = maxMessageBytes
-	} else {
-		if maxMessageBytes < options.MaxMessageBytes {
-			options.MaxMessageBytes = maxMessageBytes
-		}
+	messageMaxBytes, err := getBrokerMaxMessageBytes(admin)
+	if err != nil {
+		log.Warn("`message.max.bytes` not found from broker's configuration, use the option `MaxMessageBytes` as default",
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
+			zap.Int("maxMessageBytes", options.MaxMessageBytes), zap.Error(err))
+		messageMaxBytes = options.MaxMessageBytes
 	}
+	options.MaxMessageBytes = messageMaxBytes
 
 	// topic not exists yet, and user does not specify the `partition-num` in the sink uri.
 	if options.PartitionNum == 0 {
 		options.PartitionNum = defaultPartitionNum
 		log.Warn("partition-num is not set, use the default partition count",
+			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
 			zap.String("topic", topic), zap.Int32("partitions", options.PartitionNum))
 	}
-	return nil
+}
+
+func getTopicMaxMessageBytes(
+	ctx context.Context,
+	admin ClusterAdminClient,
+	topic string,
+) (int, error) {
+	raw, err := getTopicConfig(
+		ctx, admin, topic,
+		TopicMaxMessageBytesConfigName,
+		BrokerMessageMaxBytesConfigName,
+	)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	maxMessageBytes, err := strconv.Atoi(raw)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	return maxMessageBytes, nil
+}
+
+func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, error) {
+	raw, err := admin.GetBrokerConfig(BrokerMessageMaxBytesConfigName)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	messageMaxBytes, err := strconv.Atoi(raw)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	return messageMaxBytes, nil
 }
 
 func validateMinInsyncReplicas(
@@ -716,7 +756,7 @@ func validateMinInsyncReplicas(
 	minInsyncReplicasStr, exists, err := minInsyncReplicasConfigGetter()
 	if err != nil {
 		// 'min.insync.replica' is invisible to us in Confluent Cloud Kafka.
-		if cerror.ErrKafkaConfigNotFound.Equal(err) {
+		if errors.ErrKafkaConfigNotFound.Equal(err) {
 			log.Warn("TiCDC cannot find `min.insync.replicas` from broker's configuration, " +
 				"please make sure that the replication factor is greater than or equal " +
 				"to the minimum number of in-sync replicas" +
@@ -741,7 +781,7 @@ func validateMinInsyncReplicas(
 			MinInsyncReplicasConfigName, configFrom)
 		log.Error(msg, zap.Int("replication-factor", replicationFactor),
 			zap.Int("min.insync.replicas", minInsyncReplicas))
-		return cerror.ErrKafkaInvalidConfig.GenWithStack(
+		return errors.ErrKafkaInvalidConfig.GenWithStack(
 			"TiCDC Kafka sink's `request.required.acks` defaults to -1, "+
 				"TiCDC cannot deliver messages when the `replication-factor` %d "+
 				"is smaller than the `min.insync.replicas` %d of %s",
diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go
index 0650d3baf2..ea5d2e7147 100644
--- a/pkg/sink/kafka/options_test.go
+++ b/pkg/sink/kafka/options_test.go
@@ -28,7 +28,6 @@ import (
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
-	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
 
@@ -158,14 +157,6 @@ func (f *kafkaAdminFixture) setMessageMaxBytes(brokerValue, topicValue string) {
 	f.topicConfig[defaultMockTopicName][TopicMaxMessageBytesConfigName] = topicValue
 }
 
-func expectedAdjustedMaxMessageBytes(configuredMaxMessageBytes, sourceMaxMessageBytes int) int {
-	sourceMaxMessageBytes -= maxMessageBytesOverhead
-	if configuredMaxMessageBytes < sourceMaxMessageBytes {
-		return configuredMaxMessageBytes
-	}
-	return sourceMaxMessageBytes
-}
-
 func (f *kafkaAdminFixture) setMinInsyncReplicas(minInsyncReplicas string) {
 	f.topicConfig[defaultMockTopicName][MinInsyncReplicasConfigName] = minInsyncReplicas
 	f.brokerConfig[MinInsyncReplicasConfigName] = minInsyncReplicas
@@ -193,6 +184,7 @@ func TestCompleteOptions(t *testing.T) {
 	require.Equal(t, int16(3), options.ReplicationFactor)
 	require.Equal(t, "2.6.0", options.Version)
 	require.Equal(t, 4096, options.MaxMessageBytes)
+	require.Equal(t, 4096, options.MaxBatchedBytes)
 	require.Equal(t, WaitForLocal, options.RequiredAcks)
 	require.Equal(t, defaultMaxRetry, options.MaxRetry)
 
@@ -289,19 +281,75 @@ func TestCompleteOptions(t *testing.T) {
 	require.Equal(t, defaultMaxRetry, options.MaxRetry)
 }
 
+func TestApplyRejectsNonPositiveMaxMessageBytes(t *testing.T) {
+	tests := []struct {
+		name        string
+		uri         string
+		configValue *int
+		expected    int
+	}{
+		{
+			name:     "zero from URI",
+			uri:      "kafka://127.0.0.1:9092/test-topic?max-message-bytes=0",
+			expected: 0,
+		},
+		{
+			name:     "negative from URI",
+			uri:      "kafka://127.0.0.1:9092/test-topic?max-message-bytes=-1",
+			expected: -1,
+		},
+		{
+			name:        "zero from sink config",
+			uri:         "kafka://127.0.0.1:9092/test-topic",
+			configValue: aws.Int(0),
+			expected:    0,
+		},
+		{
+			name:        "negative from sink config",
+			uri:         "kafka://127.0.0.1:9092/test-topic",
+			configValue: aws.Int(-1),
+			expected:    -1,
+		},
+	}
+
+	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sinkURI, err := url.Parse(test.uri)
+			require.NoError(t, err)
+
+			sinkConfig := config.GetDefaultReplicaConfig().Sink
+			if test.configValue != nil {
+				sinkConfig.KafkaConfig = &config.KafkaConfig{
+					MaxMessageBytes: test.configValue,
+				}
+			}
+
+			options := NewOptions()
+			err = options.Apply(changefeedID, sinkURI, sinkConfig)
+			require.ErrorContains(
+				t, err, fmt.Sprintf("invalid max-message-bytes %d", test.expected))
+			errCode, ok := errors.RFCCode(err)
+			require.True(t, ok)
+			require.Equal(t, errors.ErrKafkaInvalidConfig.RFCCode(), errCode)
+		})
+	}
+}
+
 func TestSetPartitionNum(t *testing.T) {
 	options := NewOptions()
-	err := options.setPartitionNum(2)
+	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
+	err := options.setPartitionNum(changefeedID, 2)
 	require.NoError(t, err)
 	require.Equal(t, int32(2), options.PartitionNum)
 
 	options.PartitionNum = 1
-	err = options.setPartitionNum(2)
+	err = options.setPartitionNum(changefeedID, 2)
 	require.NoError(t, err)
 	require.Equal(t, int32(1), options.PartitionNum)
 
 	options.PartitionNum = 3
-	err = options.setPartitionNum(2)
+	err = options.setPartitionNum(changefeedID, 2)
 	require.True(t, errors.ErrKafkaInvalidPartitionNum.Equal(err))
 }
 
@@ -375,13 +423,13 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t
 		configuredMaxMessageBytes func(*kafkaAdminFixture) int
 	}{
 		{
-			name: "keeps configured value below broker limit",
+			name: "uses broker limit when configured value is below broker",
 			configuredMaxMessageBytes: func(*kafkaAdminFixture) int {
 				return 1024
 			},
 		},
 		{
-			name: "uses broker limit when configured value is within overhead",
+			name: "uses broker limit when configured value is below broker by one byte",
 			configuredMaxMessageBytes: func(f *kafkaAdminFixture) int {
 				return f.brokerMessageMaxBytes() - 1
 			},
@@ -395,7 +443,7 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t
 	}
 
 	topicName := "test-topic"
-
+	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			adminFixture := newKafkaAdminFixture(t)
@@ -408,23 +456,35 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t
 			err := adminClient.CreateTopic(detail, false)
 			require.NoError(t, err)
 
+			configuredMaxMessageBytes := test.configuredMaxMessageBytes(adminFixture)
+			sinkURI, err := url.Parse(fmt.Sprintf(
+				"kafka://127.0.0.1:9092/%s?max-message-bytes=%d",
+				topicName, configuredMaxMessageBytes,
+			))
+			require.NoError(t, err)
+
 			options := NewOptions()
-			options.BrokerEndpoints = []string{"127.0.0.1:9092"}
-			options.MaxMessageBytes = test.configuredMaxMessageBytes(adminFixture)
-			expectedMaxMessageBytes := expectedAdjustedMaxMessageBytes(
-				options.MaxMessageBytes,
-				adminFixture.brokerMessageMaxBytes(),
-			)
+			err = options.Apply(changefeedID, sinkURI, config.GetDefaultReplicaConfig().Sink)
+			require.NoError(t, err)
+			require.Equal(t, configuredMaxMessageBytes, options.MaxMessageBytes)
+			require.Equal(t, configuredMaxMessageBytes, options.MaxBatchedBytes)
+			expectedProducerLimit := adminFixture.brokerMessageMaxBytes()
 
 			ctx := context.Background()
-			err = adjustOptions(ctx, adminClient, options, topicName)
+			err = adjustOptions(ctx, changefeedID, adminClient, options, topicName)
 			require.NoError(t, err)
 
 			saramaConfig, err := newSaramaConfig(ctx, options)
 			require.NoError(t, err)
 
-			require.Equal(t, expectedMaxMessageBytes, options.MaxMessageBytes)
-			require.Equal(t, expectedMaxMessageBytes, saramaConfig.Producer.MaxMessageBytes)
+			require.NotEqual(t, configuredMaxMessageBytes, options.MaxMessageBytes)
+			require.Equal(t, expectedProducerLimit, options.MaxMessageBytes)
+			require.Equal(
+				t,
+				min(configuredMaxMessageBytes, expectedProducerLimit),
+				options.MaxBatchedBytes,
+			)
+			require.Equal(t, expectedProducerLimit, saramaConfig.Producer.MaxMessageBytes)
 		})
 	}
 }
@@ -439,10 +499,12 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) {
 	// Report an error if the replication-factor is less than min.insync.replicas
 	// when the topic does not exist.
 	adminFixture.setMinInsyncReplicas("2")
+	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
 
 	ctx := context.Background()
 	err := adjustOptions(
 		ctx,
+		changefeedID,
 		adminClient,
 		options,
 		"create-new-fail-invalid-min-insync-replicas",
@@ -456,7 +518,7 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) {
 	// topic not exist, and `min.insync.replicas` not found in broker's configuration
 	adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName)
 	topicName := "no-topic-no-min-insync-replicas"
-	err = adjustOptions(ctx, adminClient, options, "no-topic-no-min-insync-replicas")
+	err = adjustOptions(ctx, changefeedID, adminClient, options, "no-topic-no-min-insync-replicas")
 	require.Nil(t, err)
 	err = adminClient.CreateTopic(&TopicDetail{
 		Name:              topicName,
@@ -475,12 +537,12 @@ func TestAdjustConfigMinInsyncReplicas(t *testing.T) {
 		NumPartitions:     3,
 	}, false)
 	require.Nil(t, err)
-	err = adjustOptions(ctx, adminClient, options, topicName)
+	err = adjustOptions(ctx, changefeedID, adminClient, options, topicName)
 	require.Nil(t, err)
 
 	// topic found, and have `min.insync.replicas`, but set to 2, larger than `replication-factor`.
 	adminFixture.setMinInsyncReplicas("2")
-	err = adjustOptions(ctx, adminClient, options, defaultMockTopicName)
+	err = adjustOptions(ctx, changefeedID, adminClient, options, defaultMockTopicName)
 	require.Regexp(t,
 		".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of topic.*",
 		errors.Cause(err),
@@ -495,10 +557,13 @@ func TestSkipAdjustConfigMinInsyncReplicasWhenRequiredAcksIsNotWailAll(t *testin
 	options.BrokerEndpoints = []string{"127.0.0.1:9092"}
 	options.RequiredAcks = WaitForLocal
 
+	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
+
 	// Do not report an error if the replication-factor is less than min.insync.replicas(1<2).
 	adminFixture.setMinInsyncReplicas("2")
 	err := adjustOptions(
 		context.Background(),
+		changefeedID,
 		adminClient,
 		options,
 		"skip-check-min-insync-replicas",
@@ -559,7 +624,7 @@ func TestConfigurationCombinations(t *testing.T) {
 			mockTopicMessageMaxBytes,
 		},
 		{
-			"new topic broker overhead below user",
+			"new topic broker below user",
 			"kafka://127.0.0.1:9092/%s?max-message-bytes=%s",
 			[]any{"not-created-topic", strconv.Itoa(1024*1024 + 1)},
 			mockBrokerMessageMaxBytes,
@@ -625,7 +690,7 @@ func TestConfigurationCombinations(t *testing.T) {
 			strconv.Itoa(config.DefaultMaxMessageBytes + 1),
 		},
 		{
-			"existing topic topic overhead below user",
+			"existing topic topic below user",
 			"kafka://127.0.0.1:9092/%s?max-message-bytes=%s",
 			[]any{defaultMockTopicName, strconv.Itoa(1024*1024 + 1)},
 			mockBrokerMessageMaxBytes,
@@ -677,6 +742,7 @@ func TestConfigurationCombinations(t *testing.T) {
 			options := NewOptions()
 			err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink)
 			require.Nil(t, err)
+			configuredMaxMessageBytes := options.MaxMessageBytes
 
 			topic, ok := a.uriParams[0].(string)
 			require.True(t, ok)
@@ -686,30 +752,16 @@ func TestConfigurationCombinations(t *testing.T) {
 			if _, exists := adminFixture.topics[topic]; exists {
 				sourceMaxMessageBytes = adminFixture.topicMaxMessageBytes(topic)
 			}
-			expectedMaxMessageBytes := expectedAdjustedMaxMessageBytes(options.MaxMessageBytes, sourceMaxMessageBytes)
-
-			err = adjustOptions(ctx, adminClient, options, topic)
-			require.Nil(t, err)
-			require.Equal(t, expectedMaxMessageBytes, options.MaxMessageBytes)
-
-			saramaConfig, err := newSaramaConfig(ctx, options)
-			require.Nil(t, err)
-			require.Equal(t, expectedMaxMessageBytes, saramaConfig.Producer.MaxMessageBytes)
-
-			encoderConfig := common.NewConfig(config.ProtocolOpen)
-			err = encoderConfig.Apply(sinkURI, &config.SinkConfig{
-				KafkaConfig: &config.KafkaConfig{
-					LargeMessageHandle: config.NewDefaultLargeMessageHandleConfig(),
-				},
-			})
-			require.Nil(t, err)
-			encoderConfig.WithMaxMessageBytes(options.MaxMessageBytes)
 
-			err = encoderConfig.Validate()
+			changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
+			err = adjustOptions(ctx, changefeedID, adminClient, options, topic)
 			require.Nil(t, err)
-
-			// producer's `MaxMessageBytes` = encoder's `MaxMessageBytes`.
-			require.Equal(t, expectedMaxMessageBytes, encoderConfig.MaxMessageBytes)
+			require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes)
+			require.Equal(
+				t,
+				min(configuredMaxMessageBytes, sourceMaxMessageBytes),
+				options.MaxBatchedBytes,
+			)
 
 			adminClient.Close()
 		})
@@ -754,6 +806,7 @@ func TestMerge(t *testing.T) {
 	require.Equal(t, int16(5), c.ReplicationFactor)
 	require.Equal(t, "3.1.2", c.Version)
 	require.Equal(t, 1024*1024, c.MaxMessageBytes)
+	require.Equal(t, 1024*1024, c.MaxBatchedBytes)
 	require.Equal(t, "gzip", c.Compression)
 	require.Equal(t, "test-id", c.ClientID)
 	require.Equal(t, true, c.AutoCreate)
@@ -835,6 +888,7 @@ func TestMerge(t *testing.T) {
 	require.Equal(t, int16(5), c.ReplicationFactor)
 	require.Equal(t, "3.1.2", c.Version)
 	require.Equal(t, 1024*1024, c.MaxMessageBytes)
+	require.Equal(t, 1024*1024, c.MaxBatchedBytes)
 	require.Equal(t, "gzip", c.Compression)
 	require.Equal(t, "test-id", c.ClientID)
 	require.Equal(t, true, c.AutoCreate)
diff --git a/pkg/sink/kafka/sarama_config.go b/pkg/sink/kafka/sarama_config.go
index b53dc47b37..d18e8b0c56 100644
--- a/pkg/sink/kafka/sarama_config.go
+++ b/pkg/sink/kafka/sarama_config.go
@@ -62,7 +62,7 @@ func newSaramaConfig(ctx context.Context, o *options) (*sarama.Config, error) {
 	config.Producer.Flush.Bytes = 0
 	config.Producer.Flush.Messages = 0
 	config.Producer.Flush.Frequency = time.Duration(0)
-	config.Producer.Flush.MaxMessages = o.MaxMessages
+	config.Producer.Flush.MaxMessages = 0
 
 	config.Net.MaxOpenRequests = 1
 	config.Net.DialTimeout = o.DialTimeout
diff --git a/pkg/sink/kafka/sarama_config_test.go b/pkg/sink/kafka/sarama_config_test.go
index d12595bfe9..89f3e004c6 100644
--- a/pkg/sink/kafka/sarama_config_test.go
+++ b/pkg/sink/kafka/sarama_config_test.go
@@ -58,6 +58,7 @@ func TestNewSaramaConfig(t *testing.T) {
 	cfg, err := newSaramaConfig(ctx, options)
 	require.NoError(t, err)
 	require.Equal(t, defaultMaxRetry, cfg.Producer.Retry.Max)
+	require.Equal(t, options.MaxMessageBytes, cfg.Producer.MaxMessageBytes)
 
 	options.EnableTLS = true
 	options.Credential = &security.Credential{
diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go
index 650f346b4c..39c69c0c0b 100644
--- a/pkg/sink/kafka/sarama_factory.go
+++ b/pkg/sink/kafka/sarama_factory.go
@@ -57,7 +57,7 @@ func NewSaramaFactory(
 		admin.Close()
 	}()
 
-	if err = adjustOptions(ctx, admin, o, o.Topic); err != nil {
+	if err = adjustOptions(ctx, changefeedID, admin, o, o.Topic); err != nil {
 		return nil, errors.Trace(err)
 	}
 
diff --git a/tests/integration_tests/_utils/kafka_topic b/tests/integration_tests/_utils/kafka_topic
new file mode 100755
index 0000000000..9f9a79a1cd
--- /dev/null
+++ b/tests/integration_tests/_utils/kafka_topic
@@ -0,0 +1,12 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+UTILITY_DIR="$CUR/../../utils/kafka_topic"
+
+if [ ! -f "$UTILITY_DIR/kafka_topic" ]; then
+	(cd "$UTILITY_DIR" && GO111MODULE=on go build)
+fi
+
+"$UTILITY_DIR/kafka_topic" "$@"
diff --git a/tests/integration_tests/canal_json_claim_check/run.sh b/tests/integration_tests/canal_json_claim_check/run.sh
index 10f822c98b..a354dc589a 100755
--- a/tests/integration_tests/canal_json_claim_check/run.sh
+++ b/tests/integration_tests/canal_json_claim_check/run.sh
@@ -18,7 +18,10 @@ function run() {
 
 	start_tidb_cluster --workdir $WORK_DIR
 
-	TOPIC_NAME="canal-json-claim-check"
+	TOPIC_NAME="canal-json-claim-check-$RANDOM"
+	CLAIM_CHECK_DIR="/tmp/canal-json-claim-check"
+	rm -rf "$CLAIM_CHECK_DIR"
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 1000
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
@@ -36,6 +39,10 @@ function run() {
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then
+		echo "claim-check did not write any file to $CLAIM_CHECK_DIR"
+		exit 1
+	fi
 
 	cleanup_process $CDC_BINARY
 }
diff --git a/tests/integration_tests/canal_json_handle_key_only/run.sh b/tests/integration_tests/canal_json_handle_key_only/run.sh
index 372dff24fe..93ae5adfba 100755
--- a/tests/integration_tests/canal_json_handle_key_only/run.sh
+++ b/tests/integration_tests/canal_json_handle_key_only/run.sh
@@ -19,6 +19,7 @@ function run() {
 	start_tidb_cluster --workdir $WORK_DIR
 
 	TOPIC_NAME="canal-json-handle-key-only-$RANDOM"
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 1000
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
diff --git a/tests/integration_tests/kafka_big_messages/conf/diff_config.toml b/tests/integration_tests/kafka_big_messages/conf/diff_config.toml
index 0082a37028..bf73beb595 100644
--- a/tests/integration_tests/kafka_big_messages/conf/diff_config.toml
+++ b/tests/integration_tests/kafka_big_messages/conf/diff_config.toml
@@ -13,7 +13,7 @@ source-instances = ["mysql1"]
 
 target-instance = "tidb0"
 
-target-check-tables = ["kafka_big_messages.test"]
+target-check-tables = ["database_name.*"]
 
 [data-sources]
 [data-sources.mysql1]
diff --git a/tests/integration_tests/kafka_big_messages/run.sh b/tests/integration_tests/kafka_big_messages/run.sh
index 0628eaa92e..7225decaaa 100755
--- a/tests/integration_tests/kafka_big_messages/run.sh
+++ b/tests/integration_tests/kafka_big_messages/run.sh
@@ -3,53 +3,198 @@
 set -eu
 
 CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
-source $CUR/../_utils/test_prepare
+source "$CUR/../_utils/test_prepare"
 WORK_DIR=$OUT_DIR/$TEST_NAME
 CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
-function run() {
-	# test kafka sink only in this case
-	if [ "$SINK_TYPE" != "kafka" ]; then
+STATE_WAIT_TIMEOUT_SECONDS=30
+STATE_CHECK_INTERVAL_SECONDS=1
+TABLE_CHECK_RETRIES=15
+BATCH_LIMIT=262144
+SMALL_TOPIC_LIMIT=524288
+LARGE_TOPIC_LIMIT=2097152
+ROW_BYTES=1048576
+SCHEMA_REGISTRY_URI=http://127.0.0.1:8088
+GENERATOR_DIR=$CUR/../../utils/gen_kafka_big_messages
+consumer_pid=""
+
+function start_schema_registry() {
+	if curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; then
 		return
 	fi
-	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
-	start_tidb_cluster --workdir $WORK_DIR
+	echo "Starting schema registry..."
+	./bin/bin/schema-registry-start -daemon ./bin/etc/schema-registry/schema-registry.properties
+	local i=0
+	while ! curl -o /dev/null -s "$SCHEMA_REGISTRY_URI"; do
+		i=$((i + 1))
+		if [ "$i" -gt 30 ]; then
+			echo "Failed to start schema registry"
+			exit 1
+		fi
+		sleep 2
+	done
+	curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "NONE"}' "$SCHEMA_REGISTRY_URI/config"
+}
+
+function build_message_generator() {
+	if [ ! -f "$GENERATOR_DIR/gen_kafka_big_messages" ]; then
+		(cd "$GENERATOR_DIR" && GO111MODULE=on go build)
+	fi
+}
+
+function kafka_sink_uri() {
+	local topic_name=$1
+	local protocol=$2
+	local extra_params=$3
+	local sink_uri="kafka://127.0.0.1:9092/${topic_name}?protocol=${protocol}&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=${BATCH_LIMIT}"
+	if [ "$extra_params" != "" ]; then
+		sink_uri="${sink_uri}&${extra_params}"
+	fi
+	echo "$sink_uri"
+}
+
+function start_kafka_consumer() {
+	local work_dir=$1
+	local sink_uri=$2
+	local schema_registry_uri=$3
+	local protocol_case=$4
+	local downstream_uri="mysql://root@${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT}/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false"
+	local args=(
+		--log-file "$work_dir/cdc_kafka_consumer.log"
+		--log-level debug
+		--upstream-uri "$sink_uri"
+		--downstream-uri "$downstream_uri"
+	)
+	if [ "$schema_registry_uri" != "" ]; then
+		args+=(--schema-registry-uri "$schema_registry_uri")
+	fi
+	if [[ "$protocol_case" == simple_* ]]; then
+		args+=(--upstream-tidb-dsn "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?")
+	fi
+
+	cdc_kafka_consumer "${args[@]}" >>"$work_dir/cdc_kafka_consumer_stdout.log" 2>&1 &
+	consumer_pid=$!
+}
+
+function stop_kafka_consumer() {
+	if [ "$consumer_pid" != "" ]; then
+		kill -9 "$consumer_pid" 2>/dev/null || true
+		wait "$consumer_pid" 2>/dev/null || true
+		consumer_pid=""
+	fi
+}
+
+function wait_changefeed_state() {
+	local pd_addr=$1
+	local changefeed_id=$2
+	local expected_state=$3
+	local expected_error=$4
+	local deadline=$((SECONDS + STATE_WAIT_TIMEOUT_SECONDS))
 
-	TOPIC_NAME="big-message-test-$RANDOM"
+	while true; do
+		if check_changefeed_state "$pd_addr" "$changefeed_id" "$expected_state" "$expected_error" ""; then
+			return
+		fi
+		if [ "$SECONDS" -ge "$deadline" ]; then
+			echo "changefeed $changefeed_id did not reach state $expected_state within ${STATE_WAIT_TIMEOUT_SECONDS}s"
+			return 1
+		fi
+		sleep "$STATE_CHECK_INTERVAL_SECONDS"
+	done
+}
 
-	# record tso before we create tables to skip the system table DDLs
-	start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1)
+function render_diff_config() {
+	local work_dir=$1
+	local database_name=$2
+	local diff_config=$3
+
+	sed -e "s/database_name/${database_name}/g" \
+		-e "s|/tmp/tidb_cdc_test/kafka_big_messages/sync_diff/output|${work_dir}/sync_diff/output|g" \
+		"$CUR/conf/diff_config.toml" >"$diff_config"
+}
 
-	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+function run_protocol_case() {
+	local protocol_case=$1
+	local protocol=$2
+	local schema_registry_uri=$3
+	local extra_params=$4
+	local topic_case=${protocol_case//_/-}
+	local topic_name="big-message-${topic_case}-${RANDOM}"
+	local changefeed_id="kafka-big-messages-${topic_case}"
+	local database_name="kafka_big_messages_${protocol_case}"
+	local work_dir="$WORK_DIR/$protocol_case"
+	local sql_file="$work_dir/test.sql"
+	local diff_config="$work_dir/diff_config.toml"
+	local pd_addr="http://${UP_PD_HOST_1}:${UP_PD_PORT_1}"
+	local sink_uri
 
-	# Use a max-message-bytes parameter that is larger than the kafka topic max message bytes.
-	# Test if TiCDC automatically uses the max-message-bytes of the topic.
-	# See: https://github.com/PingCAP-QE/ci/blob/ddde195ebf4364a0028d53405d1194aa37a4d853/jenkins/pipelines/ci/ticdc/cdc_ghpr_kafka_integration_test.groovy#L178
-	# Use a topic that has already been created.
-	# See: https://github.com/PingCAP-QE/ci/blob/ddde195ebf4364a0028d53405d1194aa37a4d853/jenkins/pipelines/ci/ticdc/cdc_ghpr_kafka_integration_test.groovy#L180
-	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=1&kafka-version=${KAFKA_VERSION}&max-message-bytes=12582912"
-	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI"
-	run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=1&version=${KAFKA_VERSION}"
+	mkdir -p "$work_dir"
+	render_diff_config "$work_dir" "$database_name" "$diff_config"
+	kafka_topic --topic "$topic_name" --max-message-bytes "$SMALL_TOPIC_LIMIT"
+	local start_ts
+	start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1")
+	sink_uri=$(kafka_sink_uri "$topic_name" "$protocol" "$extra_params")
 
-	echo "Starting generate kafka big messages..."
-	cd $CUR/../../utils/gen_kafka_big_messages
-	if [ ! -f ./gen_kafka_big_messages ]; then
-		GO111MODULE=on go build
+	if [ "$schema_registry_uri" != "" ]; then
+		cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" --schema-registry="$schema_registry_uri" -c "$changefeed_id"
+	else
+		cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$sink_uri" -c "$changefeed_id"
 	fi
-	# Generate data larger than kafka broker max.message.bytes. We can send this data correctly.
-	./gen_kafka_big_messages --row-count=15 --sql-file-path=$CUR/test.sql
+	start_kafka_consumer "$work_dir" "$sink_uri" "$schema_registry_uri" "$protocol_case"
+	wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null"
+
+	"$GENERATOR_DIR/gen_kafka_big_messages" --row-bytes="$ROW_BYTES" --row-count=1 --database-name="$database_name" --table-name=test --sql-file-path="$sql_file"
+	run_sql_file "$sql_file" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	run_sql "CREATE TABLE ${database_name}.finish_mark(id INT PRIMARY KEY)" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+
+	# The encoded row is larger than the topic limit, so the changefeed must
+	# enter the retryable warning state with ErrMessageTooLarge.
+	wait_changefeed_state "$pd_addr" "$changefeed_id" "warning" "ErrMessageTooLarge"
+
+	# Only increase Kafka's topic limit. TiCDC must recreate the sink, read the
+	# new limit, and resume without updating, pausing, or resuming the changefeed.
+	kafka_topic --topic "$topic_name" --max-message-bytes "$LARGE_TOPIC_LIMIT" --alter
+	wait_changefeed_state "$pd_addr" "$changefeed_id" "normal" "null"
+	check_table_exists "${database_name}.finish_mark" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT" "$TABLE_CHECK_RETRIES"
+	check_sync_diff "$work_dir" "$diff_config"
+
+	cdc_cli_changefeed remove -c "$changefeed_id"
+	stop_kafka_consumer
+}
+
+function run() {
+	# Test Kafka sink only in this case.
+	if [ "$SINK_TYPE" != "kafka" ]; then
+		return
+	fi
+
+	local cases=(
+		"canal_json|canal-json||enable-tidb-extension=true"
+		"open_protocol|open-protocol||"
+		"simple_json|simple||"
+		"simple_avro|simple||encoding-format=avro"
+		"avro|avro|$SCHEMA_REGISTRY_URI|enable-tidb-extension=true&avro-enable-watermark=true&avro-decimal-handling-mode=string&avro-bigint-unsigned-handling-mode=string"
+	)
+
+	rm -rf "$WORK_DIR" && mkdir -p "$WORK_DIR"
+	start_schema_registry
+	build_message_generator
+	start_tidb_cluster --workdir "$WORK_DIR"
+	run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY"
 
-	run_sql_file $CUR/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	table="kafka_big_messages.test"
-	check_table_exists $table ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	local case_entry
+	for case_entry in "${cases[@]}"; do
+		local protocol_case protocol schema_registry_uri extra_params
+		IFS='|' read -r protocol_case protocol schema_registry_uri extra_params <<<"$case_entry"
+		run_protocol_case "$protocol_case" "$protocol" "$schema_registry_uri" "$extra_params"
+	done
 
-	cleanup_process $CDC_BINARY
+	cleanup_process "$CDC_BINARY"
 }
 
-trap 'stop_test $WORK_DIR' EXIT
-run $*
-check_logs $WORK_DIR
+trap 'stop_kafka_consumer; stop_test "$WORK_DIR"' EXIT
+run "$@"
+check_logs "$WORK_DIR"
 echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/kafka_simple_claim_check/data/data.sql b/tests/integration_tests/kafka_simple_claim_check/data/data.sql
index 2730d96d71..c753734f2e 100644
--- a/tests/integration_tests/kafka_simple_claim_check/data/data.sql
+++ b/tests/integration_tests/kafka_simple_claim_check/data/data.sql
@@ -1,4 +1,6 @@
 use test;
+-- Keep the encoded row larger than max-message-bytes after Snappy compression,
+-- so this case exercises claim-check instead of sending the full row to Kafka.
 insert into t values (
     1,
     1, 2, 3, 4, 5,
@@ -7,7 +9,7 @@ insert into t values (
     3.1415, 2.7182, 8000, 179394.233,
     '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020',
     '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A',
-    x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
+    x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024),
     '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
     'b', 'b,c', b'1000001', '{
         "key1": "value1",
@@ -28,7 +30,7 @@ insert into t values (
      3.1415, 2.7182, 8000, 179394.233,
      '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020',
      '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A',
-     x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
+     x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024),
      '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
      'b', 'b,c', b'1000001', '{
         "key1": "value1",
diff --git a/tests/integration_tests/kafka_simple_claim_check/run.sh b/tests/integration_tests/kafka_simple_claim_check/run.sh
index c8414aee56..a0ff8de268 100755
--- a/tests/integration_tests/kafka_simple_claim_check/run.sh
+++ b/tests/integration_tests/kafka_simple_claim_check/run.sh
@@ -24,6 +24,8 @@ function run() {
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="kafka-simple-claim-check-$RANDOM"
+	CLAIM_CHECK_DIR="/tmp/kafka-simple-claim-check"
+	rm -rf "$CLAIM_CHECK_DIR"
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
@@ -37,6 +39,7 @@ function run() {
 
 	cdc_cli_changefeed pause -c ${changefeed_id}
 
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 --alter
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&max-message-bytes=2048"
 	cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm
 	cdc_cli_changefeed resume -c ${changefeed_id}
@@ -48,6 +51,10 @@ function run() {
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then
+		echo "claim-check did not write any file to $CLAIM_CHECK_DIR"
+		exit 1
+	fi
 
 	cleanup_process $CDC_BINARY
 }
diff --git a/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql b/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql
index 2730d96d71..c753734f2e 100644
--- a/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql
+++ b/tests/integration_tests/kafka_simple_claim_check_avro/data/data.sql
@@ -1,4 +1,6 @@
 use test;
+-- Keep the encoded row larger than max-message-bytes after Snappy compression,
+-- so this case exercises claim-check instead of sending the full row to Kafka.
 insert into t values (
     1,
     1, 2, 3, 4, 5,
@@ -7,7 +9,7 @@ insert into t values (
     3.1415, 2.7182, 8000, 179394.233,
     '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020',
     '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A',
-    x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
+    x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024),
     '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
     'b', 'b,c', b'1000001', '{
         "key1": "value1",
@@ -28,7 +30,7 @@ insert into t values (
      3.1415, 2.7182, 8000, 179394.233,
      '2020-02-20', '2020-02-20 02:20:20', '2020-02-20 02:20:20', '02:20:20', '2020',
      '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A', '89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A89504E470D0A1A0A',
-     x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
+     x'89504E470D0A1A0A', RANDOM_BYTES(1024), RANDOM_BYTES(1024), RANDOM_BYTES(1024),
      '89504E470D0A1A0A', '89504E470D0A1A0A', x'89504E470D0A1A0A', x'89504E470D0A1A0A',
      'b', 'b,c', b'1000001', '{
         "key1": "value1",
diff --git a/tests/integration_tests/kafka_simple_claim_check_avro/run.sh b/tests/integration_tests/kafka_simple_claim_check_avro/run.sh
index 259d526d59..ac3626634e 100755
--- a/tests/integration_tests/kafka_simple_claim_check_avro/run.sh
+++ b/tests/integration_tests/kafka_simple_claim_check_avro/run.sh
@@ -24,6 +24,8 @@ function run() {
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="kafka-simple-claim-check-avro-$RANDOM"
+	CLAIM_CHECK_DIR="/tmp/kafka-simple-avro-claim-check"
+	rm -rf "$CLAIM_CHECK_DIR"
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
@@ -37,6 +39,7 @@ function run() {
 
 	cdc_cli_changefeed pause -c ${changefeed_id}
 
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048 --alter
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&encoding-format=avro&max-message-bytes=2048"
 	cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm
 	cdc_cli_changefeed resume -c ${changefeed_id}
@@ -48,6 +51,10 @@ function run() {
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then
+		echo "claim-check did not write any file to $CLAIM_CHECK_DIR"
+		exit 1
+	fi
 
 	cleanup_process $CDC_BINARY
 }
diff --git a/tests/integration_tests/kafka_simple_handle_key_only/run.sh b/tests/integration_tests/kafka_simple_handle_key_only/run.sh
index 32f7ecc6e3..e7b9f38884 100755
--- a/tests/integration_tests/kafka_simple_handle_key_only/run.sh
+++ b/tests/integration_tests/kafka_simple_handle_key_only/run.sh
@@ -36,6 +36,7 @@ function run() {
 
 	cdc_cli_changefeed pause -c ${changefeed_id}
 
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 700 --alter
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&max-message-bytes=700"
 	cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm
 	cdc_cli_changefeed resume -c ${changefeed_id}
diff --git a/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh b/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh
index 717d3924d3..94ea60d97b 100755
--- a/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh
+++ b/tests/integration_tests/kafka_simple_handle_key_only_avro/run.sh
@@ -36,6 +36,7 @@ function run() {
 
 	cdc_cli_changefeed pause -c ${changefeed_id}
 
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 650 --alter
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=simple&encoding-format=avro&max-message-bytes=650"
 	cdc_cli_changefeed update -c ${changefeed_id} --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml" --no-confirm
 	cdc_cli_changefeed resume -c ${changefeed_id}
diff --git a/tests/integration_tests/open_protocol_claim_check/data/data.sql b/tests/integration_tests/open_protocol_claim_check/data/data.sql
index 14ae2db77e..dbacc55cf2 100644
--- a/tests/integration_tests/open_protocol_claim_check/data/data.sql
+++ b/tests/integration_tests/open_protocol_claim_check/data/data.sql
@@ -93,6 +93,7 @@ insert into t values (
 );
 
 update t set c_float = 3.1415, c_double = 2.7182, c_decimal = 8000, c_decimal_2 = 179394.233 where id = 2;
+update t set c_longblob = concat(random_bytes(1024), random_bytes(1024), random_bytes(1024)) where id = 2;
 
 begin;
 
diff --git a/tests/integration_tests/open_protocol_claim_check/run.sh b/tests/integration_tests/open_protocol_claim_check/run.sh
index 2b262fd3df..7f9f94e3be 100755
--- a/tests/integration_tests/open_protocol_claim_check/run.sh
+++ b/tests/integration_tests/open_protocol_claim_check/run.sh
@@ -18,7 +18,10 @@ function run() {
 
 	start_tidb_cluster --workdir $WORK_DIR
 
-	TOPIC_NAME="open-protocol-claim-check"
+	TOPIC_NAME="open-protocol-claim-check-$RANDOM"
+	CLAIM_CHECK_DIR="/tmp/open-protocol-claim-check"
+	rm -rf "$CLAIM_CHECK_DIR"
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
@@ -40,6 +43,10 @@ function run() {
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	if ! find "$CLAIM_CHECK_DIR" -type f -print -quit | grep -q .; then
+		echo "claim-check did not write any file to $CLAIM_CHECK_DIR"
+		exit 1
+	fi
 
 	cleanup_process $CDC_BINARY
 }
diff --git a/tests/integration_tests/open_protocol_handle_key_only/data/data.sql b/tests/integration_tests/open_protocol_handle_key_only/data/data.sql
index 2977b9aa12..2c79413baf 100644
--- a/tests/integration_tests/open_protocol_handle_key_only/data/data.sql
+++ b/tests/integration_tests/open_protocol_handle_key_only/data/data.sql
@@ -93,6 +93,7 @@ insert into t values (
 );
 
 update t set c_float = 3.1415, c_double = 2.7182, c_decimal = 8000, c_decimal_2 = 179394.233 where id = 2;
+update t set c_longblob = concat(random_bytes(1024), random_bytes(1024), random_bytes(1024)) where id = 2;
 
 create table finish_mark
 (
diff --git a/tests/integration_tests/open_protocol_handle_key_only/run.sh b/tests/integration_tests/open_protocol_handle_key_only/run.sh
index a416274e1d..01aab001dc 100755
--- a/tests/integration_tests/open_protocol_handle_key_only/run.sh
+++ b/tests/integration_tests/open_protocol_handle_key_only/run.sh
@@ -19,6 +19,7 @@ function run() {
 	start_tidb_cluster --workdir $WORK_DIR
 
 	TOPIC_NAME="open-protocol-handle-key-only-$RANDOM"
+	kafka_topic --topic "$TOPIC_NAME" --max-message-bytes 2048
 
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
diff --git a/tests/utils/kafka_topic/main.go b/tests/utils/kafka_topic/main.go
new file mode 100644
index 0000000000..6227492cf9
--- /dev/null
+++ b/tests/utils/kafka_topic/main.go
@@ -0,0 +1,68 @@
+// 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 main
+
+import (
+	"flag"
+	"log"
+	"strconv"
+	"strings"
+
+	"github.com/IBM/sarama"
+)
+
+func main() {
+	brokers := flag.String("brokers", "127.0.0.1:9092", "Comma-separated Kafka broker addresses.")
+	topic := flag.String("topic", "", "Kafka topic name.")
+	maxMessageBytes := flag.Int("max-message-bytes", 0, "Topic max.message.bytes value.")
+	alter := flag.Bool("alter", false, "Alter an existing topic instead of creating it.")
+	flag.Parse()
+
+	if *topic == "" {
+		log.Fatal("topic must not be empty")
+	}
+	if *maxMessageBytes <= 0 {
+		log.Fatal("max-message-bytes must be greater than zero")
+	}
+
+	value := strconv.Itoa(*maxMessageBytes)
+	config := sarama.NewConfig()
+	config.ClientID = "ticdc-integration-test-kafka-topic"
+	admin, err := sarama.NewClusterAdmin(strings.Split(*brokers, ","), config)
+	if err != nil {
+		log.Fatalf("create Kafka admin client: %v", err)
+	}
+	defer func() {
+		if err := admin.Close(); err != nil {
+			log.Printf("close Kafka admin client: %v", err)
+		}
+	}()
+
+	configEntries := map[string]*string{"max.message.bytes": &value}
+	if *alter {
+		if err := admin.AlterConfig(sarama.TopicResource, *topic, configEntries, false); err != nil {
+			log.Fatalf("alter Kafka topic %s: %v", *topic, err)
+		}
+		return
+	}
+
+	detail := &sarama.TopicDetail{
+		NumPartitions:     1,
+		ReplicationFactor: 1,
+		ConfigEntries:     configEntries,
+	}
+	if err := admin.CreateTopic(*topic, detail, false); err != nil {
+		log.Fatalf("create Kafka topic %s: %v", *topic, err)
+	}
+}