cdc(logging): remove some verbose and useless logs (pingcap#12604)

3AceShowHand · web-flow · commit e33ff6c907b6 · 2026-04-17T02:40:03.000Z
close pingcap#11948
diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
@@ -224,7 +224,7 @@ func (c *captureImpl) reset(ctx context.Context) (*vars.GlobalVars, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	log.Info("reset session successfully", zap.Any("session", sess))
+	log.Info("reset session successfully", zap.Int64("leaseID", int64(lease.ID)))
 
 	c.captureMu.Lock()
 	defer c.captureMu.Unlock()
@@ -291,7 +291,11 @@ func (c *captureImpl) reset(ctx context.Context) (*vars.GlobalVars, error) {
 	c.processorManager = c.newProcessorManager(
 		c.info, c.upstreamManager, &c.liveness, c.config.Debug.Scheduler, globalVars)
 
-	log.Info("capture initialized", zap.Any("capture", c.info))
+	log.Info("capture initialized",
+		zap.String("captureID", c.info.ID),
+		zap.String("advertiseAddr", c.info.AdvertiseAddr),
+		zap.String("version", c.info.Version),
+		zap.String("gitHash", c.info.GitHash))
 	return globalVars, nil
 }
 
diff --git a/cdc/kv/shared_client.go b/cdc/kv/shared_client.go
@@ -824,18 +824,14 @@ func (s *SharedClient) handleResolveLockTasks(ctx context.Context) error {
 }
 
 func (s *SharedClient) logSlowRegions(ctx context.Context) error {
-	ticker := time.NewTicker(30 * time.Second)
+	ticker := time.NewTicker(5 * time.Minute)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		case <-ticker.C:
 		}
-		log.Info("event feed starts to check locked regions",
-			zap.String("namespace", s.changefeed.Namespace),
-			zap.String("changefeed", s.changefeed.ID))
-
 		currTime := s.pdClock.CurrentTime()
 		s.totalSpans.RLock()
 		var slowInitializeRegionCount int
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
@@ -218,8 +218,7 @@ func (info *ChangeFeedInfo) NeedBlockGC() bool {
 
 func (info *ChangeFeedInfo) isFailedByGC() bool {
 	if info.Error == nil {
-		log.Panic("changefeed info is not consistent",
-			zap.Any("state", info.State), zap.Any("error", info.Error))
+		log.Panic("changefeed info is not consistent", zap.Any("state", info.State))
 	}
 	return cerror.IsChangefeedGCFastFailErrorCode(errors.RFCErrorCode(info.Error.Code))
 }
@@ -381,9 +380,6 @@ func (info *ChangeFeedInfo) RmUnusedFields() {
 }
 
 func (info *ChangeFeedInfo) rmMQOnlyFields() {
-	log.Info("since the downstream is not a MQ, remove MQ only fields",
-		zap.String("namespace", info.Namespace),
-		zap.String("changefeed", info.ID))
 	info.Config.Sink.DispatchRules = nil
 	info.Config.Sink.SchemaRegistry = nil
 	info.Config.Sink.EncoderConcurrency = nil
diff --git a/cdc/model/sink.go b/cdc/model/sink.go
@@ -1193,7 +1193,7 @@ func trySplitAndSortUpdateEvent(
 	split := false
 	for _, e := range events {
 		if e == nil {
-			log.Warn("skip emit nil event", zap.Any("event", e))
+			log.Warn("skip emit nil event")
 			continue
 		}
 
diff --git a/cdc/scheduler/internal/v3/transport/transport.go b/cdc/scheduler/internal/v3/transport/transport.go
@@ -159,13 +159,6 @@ func (t *p2pTransport) Send(
 			return errors.Trace(err)
 		}
 	}
-
-	if len(msgs) != 0 {
-		log.Debug("schedulerv3: all messages sent",
-			zap.String("namespace", t.changefeed.Namespace),
-			zap.String("changefeed", t.changefeed.ID),
-			zap.Int("len", len(msgs)))
-	}
 	return nil
 }
 
diff --git a/cdc/sink/dmlsink/cloudstorage/encoding_worker.go b/cdc/sink/dmlsink/cloudstorage/encoding_worker.go
@@ -17,10 +17,8 @@ import (
 	"sync/atomic"
 
 	"github.com/pingcap/errors"
-	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/sink/codec"
-	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -52,10 +50,6 @@ func newEncodingWorker(
 }
 
 func (w *encodingWorker) run(ctx context.Context) error {
-	log.Debug("encoding worker started", zap.Int("workerID", w.id),
-		zap.String("namespace", w.changeFeedID.Namespace),
-		zap.String("changefeed", w.changeFeedID.ID))
-
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
 		for {
diff --git a/go.mod b/go.mod
@@ -439,7 +439,7 @@ require (
 	sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 // indirect
 )
 
-replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20250415
+replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20251202-x
 
 // Fix https://github.com/pingcap/tiflow/issues/4961
 replace github.com/benbjohnson/clock v1.3.5 => github.com/benbjohnson/clock v1.1.0
diff --git a/go.sum b/go.sum
@@ -888,8 +888,8 @@ github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9 h1:qG9BSvlWFEE5otQGa
 github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA=
 github.com/pingcap/metering_sdk v0.0.0-20251110022152-dac449ac5389 h1:bqbE3bwFSrUDSiN5M4EG+IXmm5eWLJnGRy/caXnxuHA=
 github.com/pingcap/metering_sdk v0.0.0-20251110022152-dac449ac5389/go.mod h1:zie1N5PRttgtqkZmRtpIDM7CuyWtvlX9LTxRd3fVSc4=
-github.com/pingcap/sarama v1.41.2-pingcap-20250415 h1:jc/31lgAuSWLh8zr3y5bL0atpBFHAjch5H1Nnlb04J0=
-github.com/pingcap/sarama v1.41.2-pingcap-20250415/go.mod h1:Kwi9CT6CuDYad3KS7HqjsbmD2DWkIKI7qI6a8PKlGb4=
+github.com/pingcap/sarama v1.41.2-pingcap-20251202-x h1:9Vi3qqyDNZxG6fnXQhpeTsnwzSBWNpMeb8o02JkL9JM=
+github.com/pingcap/sarama v1.41.2-pingcap-20251202-x/go.mod h1:xdpu7sd6OE1uxNdjYTSKUfY8FaKkJES9/+EyjSgiGQk=
 github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5 h1:T4pXRhBflzDeAhmOQHNPRRogMYxP13V7BkYw3ZsoSfE=
 github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5/go.mod h1:rlimy0GcTvjiJqvD5mXTRr8O2eNZPBrcUgiWVYp9530=
 github.com/pingcap/tidb v1.1.0-beta.0.20260325043212-0c4df2e19ecc h1:2Ateg1PUQqozi+TtbL+Bx62qn4IpQ48VTOpqCh2d3AI=
diff --git a/pkg/config/sink.go b/pkg/config/sink.go
@@ -238,7 +238,6 @@ func (s *SinkConfig) ShouldSendAllBootstrapAtStart() bool {
 		return false
 	}
 	should := s.ShouldSendBootstrapMsg() && util.GetOrZero(s.SendAllBootstrapAtStart)
-	log.Info("should send all bootstrap at start", zap.Bool("should", should))
 	return should
 }
 
@@ -845,10 +844,6 @@ func (s *SinkConfig) validateAndAdjustSinkURI(sinkURI *url.URL) error {
 		return err
 	}
 
-	log.Info("succeed to parse parameter from sink uri",
-		zap.String("protocol", util.GetOrZero(s.Protocol)),
-		zap.String("txnAtomicity", string(util.GetOrZero(s.TxnAtomicity))))
-
 	// Check that protocol config is compatible with the scheme.
 	if sink.IsMySQLCompatibleScheme(sinkURI.Scheme) && s.Protocol != nil {
 		return cerror.ErrSinkURIInvalid.GenWithStackByArgs(fmt.Sprintf("protocol %s "+
diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go
@@ -279,9 +279,7 @@ func (m *migrator) migrate(ctx context.Context, etcdNoMetaVersion bool, oldVersi
 		log.Error("update meta version failed, etcd meta data migration failed", zap.Error(err))
 		return cerror.WrapError(cerror.ErrEtcdMigrateFailed, err)
 	}
-	log.Info("etcd data migration successful")
 	cleanOldData(ctx, m.cli.GetEtcdClient())
-	log.Info("clean old etcd data successful")
 	return nil
 }
 
diff --git a/pkg/sink/pulsar/factory.go b/pkg/sink/pulsar/factory.go
@@ -131,6 +131,5 @@ func setupAuthentication(config *config.PulsarConfig) (bool, pulsar.Authenticati
 func NewMockCreatorFactory(config *config.PulsarConfig, changefeedID model.ChangeFeedID,
 	sinkConfig *config.SinkConfig,
 ) (pulsar.Client, error) {
-	log.Info("mock pulsar client factory created", zap.Any("changfeedID", changefeedID))
 	return nil, nil
 }
diff --git a/pkg/txnutil/gc/gc_manager.go b/pkg/txnutil/gc/gc_manager.go
@@ -90,9 +90,6 @@ func (m *gcManager) TryUpdateGCSafePoint(
 	failpoint.Inject("InjectActualGCSafePoint", func(val failpoint.Value) {
 		actual = uint64(val.(int))
 	})
-	if actual == checkpointTs {
-		log.Info("update gc safe point success", zap.Uint64("gcSafePointTs", checkpointTs))
-	}
 	if actual > checkpointTs {
 		log.Warn("update gc safe point failed, the gc safe point is larger than checkpointTs",
 			zap.Uint64("actual", actual), zap.Uint64("checkpointTs", checkpointTs))
diff --git a/pkg/txnutil/lock_resolver.go b/pkg/txnutil/lock_resolver.go
@@ -16,17 +16,14 @@ package txnutil
 import (
 	"bytes"
 	"context"
-	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
-	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	tikverr "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/txnkv"
-	"go.uber.org/zap"
 )
 
 // LockResolver resolves lock in the given region.
@@ -52,26 +49,6 @@ func NewLockerResolver(
 const scanLockLimit = 1024
 
 func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) (err error) {
-	var totalLocks []*txnkv.Lock
-
-	start := time.Now()
-
-	defer func() {
-		// Only log when there are locks or error to avoid log flooding.
-		if len(totalLocks) != 0 || err != nil {
-			cost := time.Since(start)
-			log.Info("resolve lock finishes",
-				zap.Uint64("regionID", regionID),
-				zap.Int("lockCount", len(totalLocks)),
-				zap.Any("locks", totalLocks),
-				zap.Uint64("maxVersion", maxVersion),
-				zap.String("namespace", r.changefeed.Namespace),
-				zap.String("changefeed", r.changefeed.ID),
-				zap.Duration("duration", cost),
-				zap.Error(err))
-		}
-	}()
-
 	// TODO test whether this function will kill active transaction
 	req := tikvrpc.NewRequest(tikvrpc.CmdScanLock, &kvrpcpb.ScanLockRequest{
 		MaxVersion: maxVersion,
@@ -130,7 +107,6 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint
 		for i := range locksInfo {
 			locks[i] = txnkv.NewLock(locksInfo[i])
 		}
-		totalLocks = append(totalLocks, locks...)
 
 		_, err1 := r.kvStorage.GetLockResolver().ResolveLocks(bo, 0, locks)
 		if err1 != nil {

Original file line number	Diff line number	Diff line change
`@@ -218,8 +218,7 @@ func (info *ChangeFeedInfo) NeedBlockGC() bool {`
`218`	`218`
`219`	`219`	`func (info *ChangeFeedInfo) isFailedByGC() bool {`
`220`	`220`	`if info.Error == nil {`
`221`		`- log.Panic("changefeed info is not consistent",`
`222`		`- zap.Any("state", info.State), zap.Any("error", info.Error))`
	`221`	`+ log.Panic("changefeed info is not consistent", zap.Any("state", info.State))`
`223`	`222`	`}`
`224`	`223`	`return cerror.IsChangefeedGCFastFailErrorCode(errors.RFCErrorCode(info.Error.Code))`
`225`	`224`	`}`
`@@ -381,9 +380,6 @@ func (info *ChangeFeedInfo) RmUnusedFields() {`
`381`	`380`	`}`
`382`	`381`
`383`	`382`	`func (info *ChangeFeedInfo) rmMQOnlyFields() {`
`384`		`- log.Info("since the downstream is not a MQ, remove MQ only fields",`
`385`		`- zap.String("namespace", info.Namespace),`
`386`		`- zap.String("changefeed", info.ID))`
`387`	`383`	`info.Config.Sink.DispatchRules = nil`
`388`	`384`	`info.Config.Sink.SchemaRegistry = nil`
`389`	`385`	`info.Config.Sink.EncoderConcurrency = nil`
Original file line number	Diff line number	Diff line change
`@@ -1193,7 +1193,7 @@ func trySplitAndSortUpdateEvent(`
`1193`	`1193`	`split := false`
`1194`	`1194`	`for _, e := range events {`
`1195`	`1195`	`if e == nil {`
`1196`		`- log.Warn("skip emit nil event", zap.Any("event", e))`
	`1196`	`+ log.Warn("skip emit nil event")`
`1197`	`1197`	`continue`
`1198`	`1198`	`}`
`1199`	`1199`
Original file line number	Diff line number	Diff line change
`@@ -159,13 +159,6 @@ func (t *p2pTransport) Send(`
`159`	`159`	`return errors.Trace(err)`
`160`	`160`	`}`
`161`	`161`	`}`
`162`		`-`
`163`		`- if len(msgs) != 0 {`
`164`		`- log.Debug("schedulerv3: all messages sent",`
`165`		`- zap.String("namespace", t.changefeed.Namespace),`
`166`		`- zap.String("changefeed", t.changefeed.ID),`
`167`		`- zap.Int("len", len(msgs)))`
`168`		`- }`
`169`	`162`	`return nil`
`170`	`163`	`}`
`171`	`164`
Original file line number	Diff line number	Diff line change
`@@ -439,7 +439,7 @@ require (`
`439`	`439`	`sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 // indirect`
`440`	`440`	`)`
`441`	`441`
`442`		`-replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20250415`
	`442`	`+replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20251202-x`
`443`	`443`
`444`	`444`	`// Fix https://github.com/pingcap/tiflow/issues/4961`
`445`	`445`	`replace github.com/benbjohnson/clock v1.3.5 => github.com/benbjohnson/clock v1.1.0`
Original file line number	Diff line number	Diff line change
`@@ -238,7 +238,6 @@ func (s *SinkConfig) ShouldSendAllBootstrapAtStart() bool {`
`238`	`238`	`return false`
`239`	`239`	`}`
`240`	`240`	`should := s.ShouldSendBootstrapMsg() && util.GetOrZero(s.SendAllBootstrapAtStart)`
`241`		`- log.Info("should send all bootstrap at start", zap.Bool("should", should))`
`242`	`241`	`return should`
`243`	`242`	`}`
`244`	`243`
`@@ -845,10 +844,6 @@ func (s SinkConfig) validateAndAdjustSinkURI(sinkURI url.URL) error {`
`845`	`844`	`return err`
`846`	`845`	`}`
`847`	`846`
`848`		`- log.Info("succeed to parse parameter from sink uri",`
`849`		`- zap.String("protocol", util.GetOrZero(s.Protocol)),`
`850`		`- zap.String("txnAtomicity", string(util.GetOrZero(s.TxnAtomicity))))`
`851`		`-`
`852`	`847`	`// Check that protocol config is compatible with the scheme.`
`853`	`848`	`if sink.IsMySQLCompatibleScheme(sinkURI.Scheme) && s.Protocol != nil {`
`854`	`849`	`return cerror.ErrSinkURIInvalid.GenWithStackByArgs(fmt.Sprintf("protocol %s "+`
Original file line number	Diff line number	Diff line change
`@@ -279,9 +279,7 @@ func (m *migrator) migrate(ctx context.Context, etcdNoMetaVersion bool, oldVersi`
`279`	`279`	`log.Error("update meta version failed, etcd meta data migration failed", zap.Error(err))`
`280`	`280`	`return cerror.WrapError(cerror.ErrEtcdMigrateFailed, err)`
`281`	`281`	`}`
`282`		`- log.Info("etcd data migration successful")`
`283`	`282`	`cleanOldData(ctx, m.cli.GetEtcdClient())`
`284`		`- log.Info("clean old etcd data successful")`
`285`	`283`	`return nil`
`286`	`284`	`}`
`287`	`285`