fix TestProcessCallBack

pkg/cdc/table_scanner.go

@@ -22,7 +22,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 
 	"go.uber.org/zap"
@@ -58,6 +57,7 @@ var GetTableDetector = func(cnUUID string) *TableDetector {
 			CallBackTableName:    make(map[string][]string),
 			SubscribedTableNames: make(map[string][]string),
 		}
+		detector.scanTableFn = detector.scanTable
 	})
 	return detector
 }
@@ -88,6 +88,8 @@ type TableDetector struct {
 	// tablename -> [taska, taskb ...]
 	SubscribedTableNames map[string][]string
 
+	scanTableFn func() error
+
 	// to make sure there is at most only one handleNewTables running, so the truncate info will not be lost
 	handling bool
 	lastMp   map[uint32]TblMap
@@ -216,19 +218,19 @@ func (s *TableDetector) scanTableLoop(ctx context.Context) {
 			s.scanAndProcess(ctx)
 		case <-retryTicker.C:
 			s.mu.Lock()
-			if s.handling || s.lastMp == nil {
-				s.mu.Unlock()
+			handling, lastMp := s.handling, s.lastMp
+			s.mu.Unlock()
+			if handling || lastMp == nil {
 				continue
 			}
-			s.mu.Unlock()
 
-			go s.processCallback(ctx, s.lastMp)
+			go s.processCallback(ctx, lastMp)
 		}
 	}
 }
 
 func (s *TableDetector) scanAndProcess(ctx context.Context) {
-	if err := s.scanTable(); err != nil {
+	if err := s.scanTableFn(); err != nil {
 		logutil.Error("CDC-TableDetector-Scan-Error", zap.Error(err))
 		return
 	}
@@ -264,10 +266,6 @@ func (s *TableDetector) processCallback(ctx context.Context, tables map[uint32]T
 }
 
 func (s *TableDetector) scanTable() error {
-	if objectio.CDCScanTableErrInjected() {
-		return moerr.NewInternalError(context.Background(), "CDC_SCANTABLE_ERR")
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	var (

pkg/cdc/table_scanner_test.go

@@ -162,13 +162,38 @@ func TestScanAndProcess(t *testing.T) {
 		exec:                 nil,
 	}
 
+	tables := map[uint32]TblMap{
+		1: {
+			"db1.tbl1": &DbTableInfo{
+				SourceDbId:      1,
+				SourceDbName:    "db1",
+				SourceTblId:     1001,
+				SourceTblName:   "tbl1",
+				SourceCreateSql: "create table tbl1 (a int)",
+				IdChanged:       false,
+			},
+		},
+	}
+	scanCount := 0
+	td.scanTableFn = func() error {
+		td.mu.Lock()
+		if scanCount%5 == 0 {
+			td.lastMp = tables
+		} else {
+			td.lastMp = nil
+		}
+		td.mu.Unlock()
+		scanCount++
+		return nil
+	}
+
 	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)
+	time.Sleep(200 * time.Millisecond)
 	defer rm()
 	td.scanAndProcess(context.Background())
 	fault.Disable()