*(ticdc): fix some major problems about pull-based-sink

cdc/entry/metrics.go

@@ -38,7 +38,7 @@ var (
 			Subsystem: "mounter",
 			Name:      "group_input_chan_size",
 			Help:      "The size of input channel of mounter group",
-		}, []string{"namespace", "changefeed", "index"})
+		}, []string{"namespace", "changefeed"})
 )
 
 // InitMetrics registers all metrics in this file

cdc/entry/mounter_group.go

@@ -15,12 +15,11 @@ package entry
 
 import (
 	"context"
-	"strconv"
-	"sync/atomic"
 	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/chann"
 	"github.com/pingcap/tiflow/pkg/filter"
 	"golang.org/x/sync/errgroup"
 )
@@ -33,13 +32,12 @@ type MounterGroup interface {
 
 type mounterGroup struct {
 	schemaStorage  SchemaStorage
-	inputCh        []chan *model.PolymorphicEvent
+	inputCh        *chann.Chann[*model.PolymorphicEvent]
 	tz             *time.Location
 	filter         filter.Filter
 	enableOldValue bool
 
 	workerNum int
-	index     uint64
 
 	changefeedID model.ChangeFeedID
 }
@@ -62,13 +60,9 @@ func NewMounterGroup(
 	if workerNum <= 0 {
 		workerNum = defaultMounterWorkerNum
 	}
-	inputCh := make([]chan *model.PolymorphicEvent, workerNum)
-	for i := 0; i < workerNum; i++ {
-		inputCh[i] = make(chan *model.PolymorphicEvent, defaultInputChanSize)
-	}
 	return &mounterGroup{
 		schemaStorage:  schemaStorage,
-		inputCh:        inputCh,
+		inputCh:        chann.New[*model.PolymorphicEvent](),
 		enableOldValue: enableOldValue,
 		filter:         filter,
 		tz:             tz,
@@ -85,29 +79,33 @@ func (m *mounterGroup) Run(ctx context.Context) error {
 	}()
 	g, ctx := errgroup.WithContext(ctx)
 	for i := 0; i < m.workerNum; i++ {
-		idx := i
 		g.Go(func() error {
-			return m.runWorker(ctx, idx)
+			return m.runWorker(ctx)
 		})
 	}
+	g.Go(func() error {
+		metrics := mounterGroupInputChanSizeGauge.WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID)
+		ticker := time.NewTicker(defaultMetricInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return errors.Trace(ctx.Err())
+			case <-ticker.C:
+				metrics.Set(float64(m.inputCh.Len()))
+			}
+		}
+	})
 	return g.Wait()
 }
 
-func (m *mounterGroup) runWorker(ctx context.Context, index int) error {
+func (m *mounterGroup) runWorker(ctx context.Context) error {
 	mounter := NewMounter(m.schemaStorage, m.changefeedID, m.tz, m.filter, m.enableOldValue)
-	rawCh := m.inputCh[index]
-	metrics := mounterGroupInputChanSizeGauge.
-		WithLabelValues(m.changefeedID.Namespace, m.changefeedID.ID, strconv.Itoa(index))
-	ticker := time.NewTicker(defaultMetricInterval)
-	defer ticker.Stop()
 	for {
-		var pEvent *model.PolymorphicEvent
 		select {
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())
-		case <-ticker.C:
-			metrics.Set(float64(len(rawCh)))
-		case pEvent = <-rawCh:
+		case pEvent := <-m.inputCh.Out():
 			if pEvent.RawKV.OpType == model.OpTypeResolved {
 				pEvent.MarkFinished()
 				continue
@@ -122,11 +120,10 @@ func (m *mounterGroup) runWorker(ctx context.Context, index int) error {
 }
 
 func (m *mounterGroup) AddEvent(ctx context.Context, event *model.PolymorphicEvent) error {
-	index := atomic.AddUint64(&m.index, 1) % uint64(m.workerNum)
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case m.inputCh[index] <- event:
+	case m.inputCh.In() <- event:
 		return nil
 	}
 }

cdc/entry/schema_storage.go

@@ -272,3 +272,38 @@ func (s *schemaStorageImpl) skipJob(job *timodel.Job) bool {
 		zap.String("changefeed", s.id.ID))
 	return !job.IsSynced() && !job.IsDone()
 }
+
+// MockSchemaStorage is for tests.
+type MockSchemaStorage struct {
+	Resolved uint64
+}
+
+// GetSnapshot implements SchemaStorage.
+func (s *MockSchemaStorage) GetSnapshot(ctx context.Context, ts uint64) (*schema.Snapshot, error) {
+	return nil, nil
+}
+
+// GetLastSnapshot implements SchemaStorage.
+func (s *MockSchemaStorage) GetLastSnapshot() *schema.Snapshot {
+	return nil
+}
+
+// HandleDDLJob implements SchemaStorage.
+func (s *MockSchemaStorage) HandleDDLJob(job *timodel.Job) error {
+	return nil
+}
+
+// AdvanceResolvedTs implements SchemaStorage.
+func (s *MockSchemaStorage) AdvanceResolvedTs(ts uint64) {
+	s.Resolved = ts
+}
+
+// ResolvedTs implements SchemaStorage.
+func (s *MockSchemaStorage) ResolvedTs() uint64 {
+	return s.Resolved
+}
+
+// DoGC implements SchemaStorage.
+func (s *MockSchemaStorage) DoGC(ts uint64) uint64 {
+	return s.Resolved
+}