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

close #8216

cdc/processor/sinkmanager/manager.go

@@ -878,7 +878,6 @@ func (m *SinkManager) Close() error {
 	}
 	m.sinkMemQuota.Close()
 	m.redoMemQuota.Close()
-	m.sinkFactory.Close()
 	m.tableSinks.Range(func(_ tablepb.Span, value interface{}) bool {
 		sink := value.(*tableSinkWrapper)
 		sink.close(m.ctx)
@@ -887,11 +886,15 @@ 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.
+	m.sinkFactory.Close()
 	log.Info("Closed sink manager",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),