use WrapError for error handling

Author: a-cong

cdc/model/sink.go

@@ -1170,7 +1170,10 @@ func (d *DDLEvent) applySinkRouting(sinkRouter *dispatcher.SinkRouter) error {
 	p := parser.New()
 	stmts, _, err := p.Parse(d.Query, d.Charset, d.Collate)
 	if err != nil {
-		return errors.Annotate(err, "failed to parse DDL for sink routing")
+		log.Error("failed to parse DDL for sink routing",
+			zap.String("query", d.Query),
+			zap.Error(err))
+		return errors.WrapError(errors.ErrDispatcherFailed, err)
 	}
 	if len(stmts) == 0 {
 		return nil
@@ -1187,7 +1190,11 @@ func (d *DDLEvent) applySinkRouting(sinkRouter *dispatcher.SinkRouter) error {
 
 	sourceTables, err := dmparser.FetchDDLTables(defaultSchema, stmts[0], conn.LCTableNamesSensitive)
 	if err != nil {
-		return errors.Annotate(err, "failed to fetch source tables for sink routing")
+		log.Error("failed to fetch source tables for sink routing",
+			zap.String("query", d.Query),
+			zap.String("defaultSchema", defaultSchema),
+			zap.Error(err))
+		return errors.WrapError(errors.ErrDispatcherFailed, err)
 	}
 
 	// Route each source table to get target schema and table in the same order
@@ -1215,7 +1222,11 @@ func (d *DDLEvent) applySinkRouting(sinkRouter *dispatcher.SinkRouter) error {
 	// Use RenameDDLTable to rewrite the query
 	routedQuery, err := dmparser.RenameDDLTable(stmts[0], targetTables)
 	if err != nil {
-		return errors.Annotate(err, "failed to rewrite DDL for sink routing")
+		log.Error("failed to rewrite DDL for sink routing",
+			zap.String("query", d.Query),
+			zap.Any("targetTables", targetTables),
+			zap.Error(err))
+		return errors.WrapError(errors.ErrDispatcherFailed, err)
 	}
 
 	d.Query = routedQuery

cdc/model/sink_test.go

@@ -1259,7 +1259,7 @@ func TestSinkRoutingError(t *testing.T) {
 		event := &DDLEvent{}
 		err := event.FromJobWithArgs(job, nil, tableInfo, "", "", router)
 		require.Error(t, err)
-		require.Contains(t, err.Error(), "failed to parse DDL for sink routing")
+		require.Contains(t, err.Error(), "ErrDispatcherFailed")
 	})
 
 	t.Run("incomplete DDL query returns error", func(t *testing.T) {
@@ -1296,7 +1296,7 @@ func TestSinkRoutingError(t *testing.T) {
 		event := &DDLEvent{}
 		err := event.FromJobWithArgs(job, nil, tableInfo, "", "", router)
 		require.Error(t, err)
-		require.Contains(t, err.Error(), "failed to parse DDL for sink routing")
+		require.Contains(t, err.Error(), "ErrDispatcherFailed")
 	})
 
 	t.Run("no error when router is nil", func(t *testing.T) {