Revert "kafka: decouple batch size from Kafka message size limit (#54…

downstreamadapter/sink/cloudstorage/encoder_group_test.go

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

downstreamadapter/sink/cloudstorage/sink.go

@@ -86,7 +86,7 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.
 	if err != nil {
 		return err
 	}
-	_, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt)
+	_, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt)
 	if err != nil {
 		return err
 	}
@@ -116,9 +116,9 @@ func New(
 	}
 	// get cloud storage file extension according to the specific protocol.
 	ext := helper.GetFileExtension(protocol)
-	// 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)
+	// 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)
 	if err != nil {
 		return nil, err
 	}

downstreamadapter/sink/helper/helper.go

@@ -50,16 +50,17 @@ func GetEncoderConfig(
 	sinkURI *url.URL,
 	protocol config.Protocol,
 	sinkConfig *config.SinkConfig,
-	maxMessageBytes int,
-	maxBatchedBytes int,
+	maxMsgBytes 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(maxMessageBytes).
-		WithMaxBatchedBytes(maxBatchedBytes).
+		WithMaxMessageBytes(maxMsgBytes).
 		WithChangefeedID(changefeedID)
 
 	tz, err := util.GetTimezone(config.GetGlobalServerConfig().TZ)

downstreamadapter/sink/kafka/helper.go

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

downstreamadapter/sink/kafka/sink.go

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

downstreamadapter/sink/kafka/sink_test.go

@@ -41,19 +41,6 @@ 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 TestSinkWorkersReturnContextError(t *testing.T) {
 	contexts := []struct {
 		name       string
@@ -197,10 +184,7 @@ func newKafkaSinkForTestWithProducers(ctx context.Context,
 	if err != nil {
 		return nil, err
 	}
-	encoderConfig, err := helper.GetEncoderConfig(
-		changefeedID, sinkURI, protocol, sinkConfig,
-		options.MaxMessageBytes, options.MaxBatchedBytes,
-	)
+	encoderConfig, err := helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, options.MaxMessageBytes)
 	if err != nil {
 		return nil, err
 	}

downstreamadapter/sink/pulsar/helper.go

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

downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

@@ -212,42 +212,50 @@ func TestEnsureTopicExistsWaitsUntilVisible(t *testing.T) {
 
 	ctrl := gomock.NewController(t)
 	adminClient := kafka.NewMockClusterAdminClient(ctrl)
-	cfg := &kafka.AutoCreateTopicConfig{
-		AutoCreate:        true,
-		PartitionNum:      2,
-		ReplicationFactor: 1,
-	}
-
-	topic := "delayed-topic"
-	gomock.InOrder(
-		adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).Return(
-			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
-			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn(
-			func(detail *kafka.TopicDetail) error {
-				require.Equal(t, &kafka.TopicDetail{
-					Name:              topic,
-					NumPartitions:     2,
-					ReplicationFactor: 1,
-				}, detail)
-				return nil
-			}),
-		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
-			map[string]kafka.TopicDetail{}, nil),
-		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).Return(
-			map[string]kafka.TopicDetail{
-				topic: {
-					Name:          topic,
+	created := false
+	postCreateDescribeCount := 0
+	adminClient.EXPECT().GetTopicsMeta([]string{"delayed-topic"}, true).Return(map[string]kafka.TopicDetail{}, nil)
+	adminClient.EXPECT().GetTopicsMeta([]string{"delayed-topic"}, false).DoAndReturn(
+		func([]string, bool) (map[string]kafka.TopicDetail, error) {
+			if !created {
+				return map[string]kafka.TopicDetail{}, nil
+			}
+			postCreateDescribeCount++
+			if postCreateDescribeCount == 1 {
+				return map[string]kafka.TopicDetail{}, nil
+			}
+			return map[string]kafka.TopicDetail{
+				"delayed-topic": {
+					Name:          "delayed-topic",
 					NumPartitions: 2,
 				},
-			}, nil),
+			}, nil
+		}).Times(3)
+	adminClient.EXPECT().CreateTopic(gomock.Any()).DoAndReturn(
+		func(detail *kafka.TopicDetail) error {
+			require.Equal(t, &kafka.TopicDetail{
+				Name:              "delayed-topic",
+				NumPartitions:     2,
+				ReplicationFactor: 1,
+			}, detail)
+			created = true
+			return nil
+		})
+
+	err := EnsureTopic(
+		context.Background(),
+		common.NewChangefeedID4Test("test", "test"),
+		"delayed-topic",
+		&kafka.AutoCreateTopicConfig{
+			AutoCreate:        true,
+			PartitionNum:      2,
+			ReplicationFactor: 1,
+		},
+		adminClient,
 	)
 
-	ctx := context.Background()
-	changefeedID := common.NewChangefeedID4Test("test", "test")
-	err := EnsureTopic(ctx, changefeedID, topic, cfg, adminClient)
 	require.NoError(t, err)
+	require.Equal(t, 2, postCreateDescribeCount)
 }
 
 func TestGetTopicManagerStartsBackgroundRefreshAfterTopicReady(t *testing.T) {

pkg/sink/codec/canal/canal_json_txn_encoder.go

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

pkg/sink/codec/common/config.go

@@ -41,12 +41,9 @@ type Config struct {
 
 	Protocol config.Protocol
 
+	// control batch behavior, only for `open-protocol` and `craft` at the moment.
 	MaxMessageBytes int
-
-	// MaxBatchedBytes controls open-protocol encoder's maximum number of bytes for a batched message.
-	MaxBatchedBytes int
-	// MaxBatchedSize controls open-protocol encoder's maximum number of events for a batched message.
-	MaxBatchSize int
+	MaxBatchSize    int
 
 	// DeleteOnlyHandleKeyColumns is true, for the delete event only output the handle key columns.
 	DeleteOnlyHandleKeyColumns bool
@@ -117,7 +114,6 @@ func NewConfig(protocol config.Protocol) *Config {
 		Protocol: protocol,
 
 		MaxMessageBytes: config.DefaultMaxMessageBytes,
-		MaxBatchedBytes: config.DefaultMaxMessageBytes,
 		MaxBatchSize:    defaultMaxBatchSize,
 
 		EnableTiDBExtension: false,
@@ -348,12 +344,6 @@ 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
@@ -423,12 +413,6 @@ func (c *Config) Validate() error {
 	if c.MaxMessageBytes <= 0 {
 		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.GenWithStack("invalid max-batch-size %d", c.MaxBatchSize)

pkg/sink/codec/common/config_test.go

@@ -33,7 +33,7 @@ func TestApplyReturnsSinkInvalidConfigForQueryBindingError(t *testing.T) {
 	require.Equal(t, errors.ErrSinkInvalidConfig.RFCCode(), errCode)
 }
 
-func TestValidateMaxBatchMessageBytes(t *testing.T) {
+func TestValidateMessageLimits(t *testing.T) {
 	tests := []struct {
 		name     string
 		adjust   func(*Config)
@@ -46,21 +46,6 @@ func TestValidateMaxBatchMessageBytes(t *testing.T) {
 			},
 			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) {

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 MaxBatchedBytes.
+// One message can contain at most MaxBatchSize events, and the total size of the message cannot exceed MaxMessageBytes.
 type batchEncoder struct {
 	messages []*common.Message
 	// buff the callback of the latest message
@@ -164,7 +164,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.MaxBatchedBytes || 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.MaxMessageBytes || d.messages[len(d.messages)-1].GetRowsCount() >= d.config.MaxBatchSize {
 		d.finalizeCallback()
 		// create a new message
 		versionHead := make([]byte, 8)

pkg/sink/codec/open/encoder_test.go

@@ -778,9 +778,7 @@ func TestEncoderMultipleMessage(t *testing.T) {
 		`insert into test.t values (3, 333)`)
 
 	ctx := context.Background()
-	codecConfig := common.NewConfig(config.ProtocolOpen).
-		WithMaxMessageBytes(1000).
-		WithMaxBatchedBytes(400)
+	codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(400)
 	encoder, err := NewBatchEncoder(codecConfig, nil)
 	require.NoError(t, err)
 
@@ -810,13 +808,11 @@ 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)
 
-	messages[0].Callback()
-	require.Equal(t, 2, count)
-	messages[1].Callback()
+	for _, message := range messages {
+		message.Callback()
+	}
+
 	require.Equal(t, 3, count)
 
 	decoder, err := NewDecoder(ctx, 0, codecConfig, nil)
@@ -889,48 +885,6 @@ 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(codecConfig, nil)
-	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++ },
-	}
-
-	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()