fix cdc scanner (#22377)

fix cdc scanner

Approved by: @daviszhen, @XuPeng-SH, @ck89119

Author: jiangxinmeng1

pkg/cdc/table_scanner.go

@@ -185,10 +185,18 @@ func (s *TableDetector) scanTableLoop(ctx context.Context) {
 	logutil.Info("CDC-TableDetector-Scan-Start")
 	defer logutil.Info("CDC-TableDetector-Scan-End")
 
-	ticker := time.NewTicker(15 * time.Second)
+	var tickerDuration, retryTickerDuration time.Duration
+	if msg, injected := objectio.CDCScanTableInjected(); injected || msg == "fast scan" {
+		tickerDuration = 1 * time.Millisecond
+		retryTickerDuration = 1 * time.Millisecond
+	} else {
+		tickerDuration = 15 * time.Second
+		retryTickerDuration = 5 * time.Second
+	}
+	ticker := time.NewTicker(tickerDuration)
 	defer ticker.Stop()
 
-	retryTicker := time.NewTicker(5 * time.Second)
+	retryTicker := time.NewTicker(retryTickerDuration)
 	defer retryTicker.Stop()
 
 	for {
@@ -205,7 +213,7 @@ func (s *TableDetector) scanTableLoop(ctx context.Context) {
 
 			s.mu.Unlock()
 
-			go s.scanAndProcess(ctx)
+			s.scanAndProcess(ctx)
 		case <-retryTicker.C:
 			s.mu.Lock()
 			if s.handling || s.lastMp == nil {
@@ -229,7 +237,7 @@ func (s *TableDetector) scanAndProcess(ctx context.Context) {
 	s.lastMp = s.Mp
 	s.mu.Unlock()
 
-	s.processCallback(ctx, s.lastMp)
+	go s.processCallback(ctx, s.lastMp)
 }
 
 func (s *TableDetector) processCallback(ctx context.Context, tables map[uint32]TblMap) {

pkg/cdc/table_scanner_test.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/util/fault"
@@ -163,6 +164,12 @@ func TestScanAndProcess(t *testing.T) {
 
 	fault.Enable()
 	objectio.SimpleInject(objectio.FJ_CDCScanTableErr)
+	rm, _ := objectio.InjectCDCScanTable("fast scan")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go td.scanTableLoop(ctx)
+	time.Sleep(2 * time.Millisecond)
+	defer rm()
 	td.scanAndProcess(context.Background())
 	fault.Disable()
 

pkg/frontend/cdc_exector.go

@@ -451,6 +451,8 @@ func (exec *CDCTaskExecutor) handleNewTables(allAccountTbls map[uint32]cdc.TblMa
 				readerInfo := reader.Info()
 				// wait the old reader to stop
 				if info.OnlyDiffinTblId(readerInfo) {
+					logutil.Infof("cdc task wait old reader to stop %s %d->%d",
+						key, readerInfo.SourceTblId, info.SourceTblId)
 					waitChan := make(chan struct{})
 					go func() {
 						defer close(waitChan)

pkg/objectio/injects.go

@@ -54,6 +54,8 @@ const (
 	FJ_CronJobsOpen = "fj/cronjobs/open"
 	FJ_CDCRecordTxn = "fj/cdc/recordtxn"
 
+	FJ_CDCScanTable = "fj/cdc/scantable"
+
 	FJ_CDCHandleSlow             = "fj/cdc/handleslow"
 	FJ_CDCHandleErr              = "fj/cdc/handleerr"
 	FJ_CDCScanTableErr           = "fj/cdc/scantableerr"
@@ -343,6 +345,11 @@ func NotifyInjected(key string) {
 	fault.TriggerFault(key)
 }
 
+func CDCScanTableInjected() (string, bool) {
+	_, sarg, injected := fault.TriggerFault(FJ_CDCScanTable)
+	return sarg, injected
+}
+
 func InjectWait(key string) (rmFault func(), err error) {
 	if err = fault.AddFaultPoint(
 		context.Background(),
@@ -452,6 +459,24 @@ func InjectGCDumpTable(msg string) (rmFault func() (bool, error), err error) {
 	return
 }
 
+func InjectCDCScanTable(msg string) (rmFault func() (bool, error), err error) {
+	if err = fault.AddFaultPoint(
+		context.Background(),
+		FJ_CDCScanTable,
+		":::",
+		"echo",
+		0,
+		msg,
+		false,
+	); err != nil {
+		return
+	}
+	rmFault = func() (ok bool, err error) {
+		return fault.RemoveFaultPoint(context.Background(), FJ_CDCScanTable)
+	}
+	return
+}
+
 func Debug19524Injected() bool {
 	_, _, injected := fault.TriggerFault(FJ_Debug19524)
 	return injected