Adds OnTableNotFound() error handler to Canal (#1105)

* Adds OnTableNotFound() error handler to Canal

* Apply suggestions from code review

Co-authored-by: Daniël van Eeden <github@myname.nl>

---------

Co-authored-by: lance6716 <lance6716@gmail.com>
Co-authored-by: Daniël van Eeden <github@myname.nl>

Author: 3 people

canal/handler.go

@@ -3,6 +3,7 @@ package canal
 import (
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/go-mysql-org/go-mysql/schema"
 )
 
 type EventHandler interface {
@@ -21,6 +22,9 @@ type EventHandler interface {
 	// You'll get the original executed query, with comments if present.
 	// It will be called before OnRow.
 	OnRowsQueryEvent(e *replication.RowsQueryEvent) error
+	// OnTableNotFound is called when a Rows Event references a table object
+	// that no longer exists.
+	OnTableNotFound(*replication.EventHeader, *replication.RowsEvent) error
 	String() string
 }
 
@@ -51,6 +55,11 @@ func (h *DummyEventHandler) OnRowsQueryEvent(*replication.RowsQueryEvent) error
 	return nil
 }
 
+// OnTableNotFound is called for row events for reference tables that are not found
+func (h *DummyEventHandler) OnTableNotFound(header *replication.EventHeader, e *replication.RowsEvent) error {
+	return schema.ErrTableNotExist
+}
+
 func (h *DummyEventHandler) String() string { return "DummyEventHandler" }
 
 // `SetEventHandler` registers the sync handler, you must register your

canal/sync.go

@@ -270,12 +270,15 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 
 	t, err := c.GetTable(schemaName, tableName)
 	if err != nil {
-		e := errors.Cause(err)
+		cause := errors.Cause(err)
 		// ignore errors below
-		if e == ErrExcludedTable || e == schema.ErrTableNotExist || e == schema.ErrMissingTableMeta {
-			err = nil
+		if cause == ErrExcludedTable || cause == schema.ErrMissingTableMeta {
+			return nil
+		}
+		// Allow handler to decide what to do when table is missing.
+		if cause == schema.ErrTableNotExist {
+			return c.eventHandler.OnTableNotFound(e.Header, ev)
 		}
-
 		return err
 	}
 	var action string