[DNM] log(ticdc): always print log from the sarama inside to make debug life easier

go.mod

@@ -417,5 +417,7 @@ replace sourcegraph.com/sourcegraph/appdash => github.com/sourcegraph/appdash v0
 
 replace sourcegraph.com/sourcegraph/appdash-data => github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
 
+replace github.com/IBM/sarama => github.com/3AceShowHand/sarama v0.0.0-20241204051647-318559e536ae
+
 // tls10server=1
 godebug tlsrsakex=1

go.sum

@@ -31,6 +31,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.39.1 h1:MvraqHKhogCOTXTlct/9C3K3+Uy2jBmFYb3/Sp6dVtY=
 cloud.google.com/go/storage v1.39.1/go.mod h1:xK6xZmxZmo+fyP7+DEF6FhNc24/JAe95OLyOHCXFH1o=
+github.com/3AceShowHand/sarama v0.0.0-20241204051647-318559e536ae h1:TZxjgE3hc0iM0GIxz1RywfrEheWz8CsSYAEXyo6EZV8=
+github.com/3AceShowHand/sarama v0.0.0-20241204051647-318559e536ae/go.mod h1:xdpu7sd6OE1uxNdjYTSKUfY8FaKkJES9/+EyjSgiGQk=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=

pkg/logutil/log.go

@@ -223,14 +223,11 @@ func initMySQLLogger() error {
 
 // initSaramaLogger hacks logger used in sarama lib
 func initSaramaLogger(level zapcore.Level) error {
-	// only available less than info level
-	if !zapcore.InfoLevel.Enabled(level) {
-		logger, err := zap.NewStdLogAt(log.L().With(zap.String("component", "sarama")), level)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		sarama.Logger = logger
+	logger, err := zap.NewStdLogAt(log.L().With(zap.String("component", "sarama")), level)
+	if err != nil {
+		return errors.Trace(err)
 	}
+	sarama.Logger = logger
 	return nil
 }
 

pkg/sink/codec/common/message.go

@@ -16,8 +16,10 @@ package common
 import (
 	"encoding/binary"
 	"encoding/json"
+	"strconv"
 	"time"
 
+	"github.com/IBM/sarama"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/tikv/client-go/v2/oracle"
@@ -50,7 +52,21 @@ type Message struct {
 // We didn't append any `Headers` when send the message, so ignore the calculations related to it.
 // If `ProducerMessage` Headers fields used, this method should also adjust.
 func (m *Message) Length() int {
-	return len(m.Key) + len(m.Value) + MaxRecordOverhead
+	headers := m.Headers()
+	var headerLen int
+	for _, header := range headers {
+		headerLen += len(header.Key) + len(header.Value) + 2*binary.MaxVarintLen32
+	}
+	return headerLen + len(m.Key) + len(m.Value) + MaxRecordOverhead
+}
+
+// Headers returns the headers of Kafka message
+func (m *Message) Headers() []sarama.RecordHeader {
+	headers := []sarama.RecordHeader{
+		{Key: []byte("commitTs"), Value: []byte(strconv.FormatUint(m.Ts, 10))},
+		{Key: []byte("table"), Value: []byte(m.GetTable())},
+	}
+	return headers
 }
 
 // PhysicalTime returns physical time part of Ts in time.Time

pkg/sink/kafka/factory.go

@@ -257,6 +257,7 @@ func (p *saramaAsyncProducer) AsyncSend(ctx context.Context, topic string, parti
 	msg := &sarama.ProducerMessage{
 		Topic:     topic,
 		Partition: partition,
+		Headers:   message.Headers(),
 		Key:       sarama.StringEncoder(message.Key),
 		Value:     sarama.ByteEncoder(message.Value),
 		Metadata:  message.Callback,
@@ -266,5 +267,6 @@ func (p *saramaAsyncProducer) AsyncSend(ctx context.Context, topic string, parti
 		return errors.Trace(ctx.Err())
 	case p.producer.Input() <- msg:
 	}
+	log.Info("async send msg", zap.Int32("partition", partition), zap.Any("commitTs", message.Ts), zap.String("table", message.GetTable()), zap.String("shcema", message.GetSchema()))
 	return nil
 }