sinkv2(cdc): fix some races of source manager (#7996)

Signed-off-by: qupeng <[email protected]>

cdc/processor/sinkmanager/redo_log_worker.go

@@ -121,12 +121,10 @@ func (w *redoWorker) handleTask(ctx context.Context, task *redoTask) (finalErr e
 			if err != nil {
 				return errors.Trace(err)
 			}
+			// Should always re-allocate space because EmitRowChangedEvents is asynchronous.
+			rows = make([]*model.RowChangedEvent, 0, 1024)
 			rowsSize = 0
 			cachedSize = 0
-			rows = rows[:0]
-			if cap(rows) > 1024 {
-				rows = make([]*model.RowChangedEvent, 0, 1024)
-			}
 		}
 		if lastTxnCommitTs > emitedCommitTs {
 			if err := w.redoManager.UpdateResolvedTs(

cdc/processor/sourcemanager/engine/factory/factory.go

@@ -41,6 +41,12 @@ const (
 	metricsCollectInterval = 15 * time.Second
 )
 
+var (
+	// Use singleton to be compatible with test cases.
+	factoryMu sync.Mutex
+	factory   *SortEngineFactory = nil
+)
+
 // SortEngineFactory is a manager to create or drop SortEngine.
 type SortEngineFactory struct {
 	// Read-only fields.
@@ -105,6 +111,10 @@ func (f *SortEngineFactory) Drop(ID model.ChangeFeedID) error {
 
 // Close will close all created engines and release all resources.
 func (f *SortEngineFactory) Close() (err error) {
+	factoryMu.Lock()
+	defer factoryMu.Unlock()
+	factory = nil
+
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
@@ -122,18 +132,21 @@ func (f *SortEngineFactory) Close() (err error) {
 
 // NewForPebble will create a SortEngineFactory for the pebble implementation.
 func NewForPebble(dir string, memQuotaInBytes uint64, cfg *config.DBConfig) *SortEngineFactory {
-	manager := &SortEngineFactory{
-		engineType:      pebbleEngine,
-		dir:             dir,
-		memQuotaInBytes: memQuotaInBytes,
-		engines:         make(map[model.ChangeFeedID]engine.SortEngine),
-		closed:          make(chan struct{}),
-		pebbleConfig:    cfg,
-		dbInitialized:   atomic.NewBool(false),
+	factoryMu.Lock()
+	defer factoryMu.Unlock()
+	if factory == nil {
+		factory = &SortEngineFactory{
+			engineType:      pebbleEngine,
+			dir:             dir,
+			memQuotaInBytes: memQuotaInBytes,
+			engines:         make(map[model.ChangeFeedID]engine.SortEngine),
+			closed:          make(chan struct{}),
+			pebbleConfig:    cfg,
+			dbInitialized:   atomic.NewBool(false),
+		}
+		factory.startMetricsCollector()
 	}
-
-	manager.startMetricsCollector()
-	return manager
+	return factory
 }
 
 func (f *SortEngineFactory) startMetricsCollector() {