processor,sink(cdc): let sink report resolved ts and do not skip buff…

…er sink flush (#3540) (#3561)

cdc/sink/manager.go

@@ -110,9 +110,12 @@ func (m *Manager) flushBackendSink(ctx context.Context) (model.Ts, error) {
 	// NOTICE: Because all table sinks will try to flush backend sink,
 	// which will cause a lot of lock contention and blocking in high concurrency cases.
 	// So here we use flushing as a lightweight lock to improve the lock competition problem.
-	if !atomic.CompareAndSwapInt64(&m.flushing, 0, 1) {
-		return m.getCheckpointTs(), nil
-	}
+	//
+	// Do not skip flushing for resolving #3503.
+	// TODO uncomment the following return.
+	// if !atomic.CompareAndSwapInt64(&m.flushing, 0, 1) {
+	// return m.getCheckpointTs(), nil
+	// }
 	m.flushMu.Lock()
 	defer func() {
 		m.flushMu.Unlock()