sinkV2(ticdc): sink manager close the sink factory at the last (#8219) (

#8227)

close #8216

cdc/processor/sinkmanager/manager.go

@@ -878,10 +878,6 @@ func (m *SinkManager) Close() error {
 	}
 	m.sinkMemQuota.Close()
 	m.redoMemQuota.Close()
-	err := m.sinkFactory.Close()
-	if err != nil {
-		return errors.Trace(err)
-	}
 	m.tableSinks.Range(func(key, value interface{}) bool {
 		sink := value.(*tableSinkWrapper)
 		sink.close(m.ctx)
@@ -890,11 +886,18 @@ func (m *SinkManager) Close() error {
 		}
 		return true
 	})
+	m.wg.Wait()
 	log.Info("All table sinks closed",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),
 		zap.Duration("cost", time.Since(start)))
-	m.wg.Wait()
+	// todo: Add a unit test to cover this,
+	// Make sure all sink workers exited before closing the sink factory.
+	// Otherwise, it would panic in the sink when you try to write some data to a closed sink.
+	err := m.sinkFactory.Close()
+	if err != nil {
+		return errors.Trace(err)
+	}
 	log.Info("Closed sink manager",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),