kafka(ticdc): fix kafka dml sink report error to sink manager (#9879) (…

…#9880) ref #9855
pingcap · Oct 18, 2023 · ef317bf · ef317bf
1 parent c27eddf
commit ef317bf
Showing 1 changed file with 16 additions and 2 deletions.
diff --git a/cdc/sinkv2/eventsink/mq/mq_dml_sink.go b/cdc/sinkv2/eventsink/mq/mq_dml_sink.go
@@ -117,10 +117,24 @@ func newSink(ctx context.Context,
 		s.alive.Unlock()
 		close(s.dead)
 
-		if err != nil && errors.Cause(err) != context.Canceled {
+		if err != nil {
+			if errors.Cause(err) == context.Canceled {
+				if s.cancelErr == nil {
+					return
+				}
+				err = s.cancelErr
+			}
 			select {
-			case <-ctx.Done():
 			case errCh <- err:
+				log.Warn("mq dml sink meet error",
+					zap.String("namespace", s.id.Namespace),
+					zap.String("changefeed", s.id.ID),
+					zap.Error(err))
+			default:
+				log.Info("mq dml sink meet error, ignored",
+					zap.String("namespace", s.id.Namespace),
+					zap.String("changefeed", s.id.ID),
+					zap.Error(err))
 			}
 		}
 	}()