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

ref #9855
pingcap · Oct 13, 2023 · 07ad2c1 · 07ad2c1
1 parent 3777731
commit 07ad2c1
Showing 1 changed file with 13 additions and 2 deletions.
diff --git a/cdc/sink/dmlsink/mq/mq_dml_sink.go b/cdc/sink/dmlsink/mq/mq_dml_sink.go
@@ -114,10 +114,21 @@ func newDMLSink(
 		s.alive.Unlock()
 		close(s.dead)
 
-		if err != nil && errors.Cause(err) != context.Canceled {
+		if err != nil {
+			if errors.Cause(err) == context.Canceled {
+				err = context.Cause(ctx)
+			}
 			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))
 			}
 		}
 	}()