sink(cdc): handle sink errors more fast and light

cdc/model/errors.go

@@ -20,7 +20,7 @@ import (
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
 
-// RunningError represents some running error from cdc components, such as processor.
+// RunningError represents some running errors from cdc components, such as processor.
 type RunningError struct {
 	Time    time.Time `json:"time"`
 	Addr    string    `json:"addr"`
@@ -32,3 +32,27 @@ type RunningError struct {
 func (r RunningError) IsChangefeedUnRetryableError() bool {
 	return cerror.IsChangefeedUnRetryableError(errors.New(r.Message + r.Code))
 }
+
+const (
+	// ComponentProcessorSink indicates the sink module in processor.
+	ComponentProcessorSink string = "processor/sink"
+	// ComponentOwnerSink indicates the sink module in owner.
+	ComponentOwnerSink string = "owner/sink"
+)
+
+// Warning is like an error, but has one difference:
+// generally an error will stop and restart a changefeed, but a warning won't.
+type Warning struct {
+	err       error
+	Component string
+}
+
+// Error implements builtin `error` interface.
+func (e Warning) Error() string {
+	return e.err.Error()
+}
+
+// NewWarning creates a Warning.
+func NewWarning(e error, component string) Warning {
+	return Warning{e, component}
+}

cdc/owner/changefeed.go

@@ -129,7 +129,11 @@ type changefeed struct {
 		filter filter.Filter,
 	) (puller.DDLPuller, error)
 
-	newSink      func(changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo, reportErr func(error)) DDLSink
+	newSink func(
+		changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
+		reportError func(err error), reportWarning func(err error),
+	) DDLSink
+
 	newScheduler func(
 		ctx cdcContext.Context, up *upstream.Upstream, epoch uint64, cfg *config.SchedulerConfig,
 	) (scheduler.Scheduler, error)
@@ -179,7 +183,10 @@ func newChangefeed4Test(
 		schemaStorage entry.SchemaStorage,
 		filter filter.Filter,
 	) (puller.DDLPuller, error),
-	newSink func(changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo, reportErr func(err error)) DDLSink,
+	newSink func(
+		changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
+		reportError func(err error), reportWarning func(err error),
+	) DDLSink,
 	newScheduler func(
 		ctx cdcContext.Context, up *upstream.Upstream, epoch uint64, cfg *config.SchedulerConfig,
 	) (scheduler.Scheduler, error),
@@ -242,6 +249,14 @@ func (c *changefeed) handleErr(ctx cdcContext.Context, err error) {
 	} else {
 		code = string(cerror.ErrOwnerUnknown.RFCCode())
 	}
+
+	switch errors.Cause(err).(type) {
+	case model.Warning:
+		// TODO(qupeng): patch it into changefeed info.
+		return
+	default:
+	}
+
 	c.feedStateManager.handleError(&model.RunningError{
 		Time:    time.Now(),
 		Addr:    contextutil.CaptureAddrFromCtx(ctx),
@@ -296,11 +311,7 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 		return errors.Trace(err)
 	default:
 	}
-	// we need to wait ddl ddlSink to be ready before we do the other things
-	// otherwise, we may cause a nil pointer panic when we try to write to the ddl ddlSink.
-	if !c.ddlSink.isInitialized() {
-		return nil
-	}
+
 	// TODO: pass table checkpointTs when we support concurrent process ddl
 	allPhysicalTables, minTableBarrierTs, barrier, err := c.ddlManager.tick(ctx, checkpointTs, nil)
 	if err != nil {
@@ -567,7 +578,13 @@ LOOP:
 		zap.String("changefeed", c.id.ID),
 	)
 
-	c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw)
+	c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw, func(err error) {
+		// TODO(qupeng): report the warning.
+		log.Info("ddlSink internal error",
+			zap.String("namespace", c.id.Namespace),
+			zap.String("changefeed", c.id.ID),
+			zap.Error(err))
+	})
 	c.ddlSink.run(cancelCtx)
 
 	c.ddlPuller, err = c.newDDLPuller(cancelCtx,
@@ -586,8 +603,7 @@ LOOP:
 		ctx.Throw(c.ddlPuller.Run(cancelCtx))
 	}()
 
-	c.downstreamObserver, err = c.newDownstreamObserver(
-		ctx, c.state.Info.SinkURI, c.state.Info.Config)
+	c.downstreamObserver, err = c.newDownstreamObserver(ctx, c.state.Info.SinkURI, c.state.Info.Config)
 	if err != nil {
 		return err
 	}
@@ -1013,8 +1029,8 @@ func (c *changefeed) tickDownstreamObserver(ctx context.Context) {
 			defer cancel()
 			if err := c.downstreamObserver.Tick(cctx); err != nil {
 				// Prometheus is not deployed, it happens in non production env.
-				if strings.Contains(err.Error(),
-					fmt.Sprintf(":%d", errno.ErrPrometheusAddrIsNotSet)) {
+				noPrometheusMsg := fmt.Sprintf(":%d", errno.ErrPrometheusAddrIsNotSet)
+				if strings.Contains(err.Error(), noPrometheusMsg) {
 					return
 				}
 				log.Warn("backend observer tick error", zap.Error(err))

cdc/owner/changefeed_test.go

@@ -150,10 +150,6 @@ func (m *mockDDLSink) close(ctx context.Context) error {
 	return nil
 }
 
-func (m *mockDDLSink) isInitialized() bool {
-	return true
-}
-
 func (m *mockDDLSink) Barrier(ctx context.Context) error {
 	return nil
 }