kafka: decouple batch size from Kafka message size limit (#5420)

downstreamadapter/sink/cloudstorage/encoder_group_test.go

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

downstreamadapter/sink/cloudstorage/sink.go

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

downstreamadapter/sink/helper/helper.go

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

downstreamadapter/sink/kafka/helper.go

@@ -102,7 +102,10 @@ func newKafkaSinkComponent(
 		return comp, protocol, 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 comp, protocol, err
 	}

downstreamadapter/sink/kafka/sink.go

@@ -88,7 +88,10 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, uri *url.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 err
 	}
@@ -253,7 +256,7 @@ func (s *sink) WriteBlockEvent(event commonEvent.BlockEvent) error {
 	case *commonEvent.DDLEvent:
 		err = s.sendDDLEvent(v)
 	default:
-		log.Error("kafka sink doesn't support this type of block event",
+		log.Error("unsupported kafka sink block event type",
 			zap.String("namespace", s.changefeedID.Keyspace()),
 			zap.String("changefeed", s.changefeedID.Name()),
 			zap.String("eventType", commonEvent.TypeToString(event.GetType())))
@@ -307,9 +310,6 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error {
 		default:
 			event, ok := s.eventChan.Get()
 			if !ok {
-				log.Info("kafka sink event channel closed",
-					zap.String("keyspace", s.changefeedID.Keyspace()),
-					zap.String("changefeed", s.changefeedID.Name()))
 				return nil
 			}
 			schema := event.TableInfo.GetSchemaName()
@@ -369,9 +369,6 @@ func (s *sink) nonBatchEncodeRun(ctx context.Context) error {
 		default:
 			event, ok := s.rowChan.Get()
 			if !ok {
-				log.Info("kafka sink event channel closed",
-					zap.String("keyspace", s.changefeedID.Keyspace()),
-					zap.String("changefeed", s.changefeedID.Name()))
 				return nil
 			}
 			if err := s.comp.encoderGroup.AddEvents(ctx, event.Key, &event.RowEvent); err != nil {
@@ -394,10 +391,6 @@ func (s *sink) batchEncodeRun(ctx context.Context) error {
 		start := time.Now()
 		msgs, err := s.batch(ctx, msgsBuf)
 		if err != nil {
-			log.Error("kafka sink batch dml events failed",
-				zap.String("keyspace", s.changefeedID.Keyspace()),
-				zap.String("changefeed", s.changefeedID.Name()),
-				zap.Error(err))
 			return err
 		}
 		if len(msgs) == 0 {
@@ -427,9 +420,6 @@ func (s *sink) batch(ctx context.Context, buffer []*commonEvent.MQRowEvent) ([]*
 	default:
 		msgs, ok := s.rowChan.GetMultipleNoGroup(buffer)
 		if !ok {
-			log.Info("kafka sink event channel closed",
-				zap.String("keyspace", s.changefeedID.Keyspace()),
-				zap.String("changefeed", s.changefeedID.Name()))
 			return nil, nil
 		}
 		buffer = buffer[:0]
@@ -461,9 +451,6 @@ func (s *sink) sendMessages(ctx context.Context) error {
 			return context.Cause(ctx)
 		case future, ok := <-outCh:
 			if !ok {
-				log.Info("kafka sink encoder's output channel closed",
-					zap.String("keyspace", s.changefeedID.Keyspace()),
-					zap.String("changefeed", s.changefeedID.Name()))
 				return nil
 			}
 			if err = future.Ready(ctx); err != nil {
@@ -473,16 +460,11 @@ func (s *sink) sendMessages(ctx context.Context) error {
 				start := time.Now()
 				if err = s.statistics.RecordBatchExecution(func() (int, int64, error) {
 					message.SetPartitionKey(future.Key.PartitionKey)
-					log.Debug("send message to kafka", zap.String("messageKey", util.RedactBytes(message.Key)), zap.String("messageValue", util.RedactBytes(message.Value)))
 					if err = s.dmlProducer.AsyncSend(
 						ctx,
 						future.Key.Topic,
 						future.Key.Partition,
 						message); err != nil {
-						log.Error("kafka sink send message failed",
-							zap.String("keyspace", s.changefeedID.Keyspace()),
-							zap.String("changefeed", s.changefeedID.Name()),
-							zap.Error(err))
 						return 0, 0, err
 					}
 					return message.GetRowsCount(), int64(message.Length()), nil
@@ -502,9 +484,10 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error {
 			return err
 		}
 		if message == nil {
-			log.Info("Skip ddl event", zap.Uint64("startTs", event.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()),
-				zap.String("query", e.Query),
-				zap.Stringer("changefeed", s.changefeedID))
+			log.Info("kafka ddl event skipped",
+				zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()),
+				zap.Uint64("startTs", e.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()),
+				zap.String("query", e.Query))
 			continue
 		}
 		codecCommon.SetDDLMessageLogInfo(message, e)
@@ -530,11 +513,11 @@ func (s *sink) sendDDLEvent(event *commonEvent.DDLEvent) error {
 		if err != nil {
 			return err
 		}
+		log.Info("kafka ddl event sent",
+			zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()),
+			zap.Uint64("startTs", e.GetStartTs()), zap.Uint64("commitTs", e.GetCommitTs()),
+			zap.String("query", e.GetDDLQuery()))
 	}
-	log.Info("kafka sink send DDL event",
-		zap.String("keyspace", s.changefeedID.Keyspace()), zap.String("changefeed", s.changefeedID.Name()),
-		zap.Any("startTs", event.GetStartTs()), zap.Any("commitTs", event.GetCommitTs()), zap.Any("event", event.GetDDLQuery()),
-		zap.String("schema", event.GetSchemaName()), zap.String("table", event.GetTableName()))
 	return nil
 }
 
@@ -567,9 +550,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 			return context.Cause(ctx)
 		case ts, ok := <-s.checkpointChan:
 			if !ok {
-				log.Warn("kafka sink checkpoint channel closed",
-					zap.String("keyspace", s.changefeedID.Keyspace()),
-					zap.String("changefeed", s.changefeedID.Name()))
 				return nil
 			}
 
@@ -622,10 +602,6 @@ func (s *sink) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaStor
 
 func (s *sink) getAllTableNames(ts uint64) []*commonEvent.SchemaTableName {
 	if s.tableSchemaStore == nil {
-		log.Warn("kafka sink table schema store is not set",
-			zap.String("keyspace", s.changefeedID.Keyspace()),
-			zap.String("changefeed", s.changefeedID.Name()),
-			zap.Uint64("ts", ts))
 		return nil
 	}
 	return s.tableSchemaStore.GetAllTableNames(ts)

downstreamadapter/sink/kafka/sink_test.go

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

downstreamadapter/sink/pulsar/helper.go

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