cdc: remove unused code and add some logs (#10458)

ref #10457

cdc/model/sink.go

@@ -18,7 +18,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"unsafe"
 
@@ -778,11 +777,6 @@ type SingleTableTxn struct {
 	StartTs  uint64
 	CommitTs uint64
 	Rows     []*RowChangedEvent
-
-	// control fields of SingleTableTxn
-	// FinishWg is a barrier txn, after this txn is received, the worker must
-	// flush cached txns and call FinishWg.Done() to mark txns have been flushed.
-	FinishWg *sync.WaitGroup
 }
 
 // GetCommitTs returns the commit timestamp of the transaction.
@@ -924,11 +918,6 @@ func (t *SingleTableTxn) Append(row *RowChangedEvent) {
 	t.Rows = append(t.Rows, row)
 }
 
-// ToWaitFlush indicates whether to wait flushing after the txn is processed or not.
-func (t *SingleTableTxn) ToWaitFlush() bool {
-	return t.FinishWg != nil
-}
-
 // TopicPartitionKey contains the topic and partition key of the message.
 type TopicPartitionKey struct {
 	Topic        string

cdc/processor/sinkmanager/metrics.go

@@ -18,17 +18,6 @@ import (
 )
 
 var (
-	// MemoryQuota indicates memory usage of a changefeed.
-	MemoryQuota = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: "ticdc",
-			Subsystem: "sinkmanager",
-			Name:      "memory_quota",
-			Help:      "memory quota of the changefeed",
-		},
-		// type includes total, used, component includes sink and redo.
-		[]string{"namespace", "changefeed", "type", "component"})
-
 	// RedoEventCache indicates redo event memory usage of a changefeed.
 	RedoEventCache = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{

cdc/sink/dmlsink/txn/mysql/mysql.go

@@ -207,7 +207,7 @@ func NewMySQLBackends(
 func (s *mysqlBackend) OnTxnEvent(event *dmlsink.TxnCallbackableEvent) (needFlush bool) {
 	s.events = append(s.events, event)
 	s.rows += len(event.Event.Rows)
-	return event.Event.ToWaitFlush() || s.rows >= s.cfg.MaxTxnRow
+	return s.rows >= s.cfg.MaxTxnRow
 }
 
 // Flush implements interface backend.

cdc/sink/dmlsink/txn/worker.go

@@ -92,7 +92,7 @@ func (w *worker) close() {
 	w.txnCh.CloseAndDrain()
 }
 
-// Run a loop.
+// Continuously get events from txnCh and call backend flush based on conditions.
 func (w *worker) runLoop() error {
 	defer func() {
 		if err := w.backend.Close(); err != nil {
@@ -148,7 +148,7 @@ func (w *worker) runLoop() error {
 }
 
 // onEvent is called when a new event is received.
-// It returns true if the event is sent to backend.
+// It returns true if it needs flush immediately.
 func (w *worker) onEvent(txn txnWithNotifier) bool {
 	w.hasPending = true
 
@@ -169,7 +169,6 @@ func (w *worker) onEvent(txn txnWithNotifier) bool {
 }
 
 // doFlush flushes the backend.
-// It returns true only if it can no longer be flushed.
 func (w *worker) doFlush(flushTimeSlice *time.Duration) error {
 	if w.hasPending {
 		start := time.Now()