From cea2577c9dc9ffe05a0aabbc7116113ed7bafe18 Mon Sep 17 00:00:00 2001
From: adamszpilewicz <57848391+adamszpilewicz@users.noreply.github.com>
Date: Tue, 3 Jan 2023 14:48:00 +0100
Subject: [PATCH 1/2] skip anonymous transactions

---
 canal/sync.go | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/canal/sync.go b/canal/sync.go
index e03d01267..11acc1260 100644
--- a/canal/sync.go
+++ b/canal/sync.go
@@ -266,6 +266,9 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 		action = DeleteAction
 	case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
 		action = UpdateAction
+	case replication.ANONYMOUS_GTID_EVENT:
+		// Skip anonymous transactions
+		return nil
 	default:
 		return errors.Errorf("%s not supported now", e.Header.EventType)
 	}

From 051030712249d592e39fee2be6a4b4a664583397 Mon Sep 17 00:00:00 2001
From: adamszpilewicz <57848391+adamszpilewicz@users.noreply.github.com>
Date: Wed, 4 Jan 2023 15:37:07 +0100
Subject: [PATCH 2/2] log anonmymous transaction

---
 canal/sync.go | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/canal/sync.go b/canal/sync.go
index 11acc1260..9f8ff803a 100644
--- a/canal/sync.go
+++ b/canal/sync.go
@@ -248,6 +248,10 @@ func (c *Canal) updateReplicationDelay(ev *replication.BinlogEvent) {
 }
 
 func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
+	if e.Header.EventType == replication.ANONYMOUS_GTID_EVENT {
+		log.Printf("==> Anonymous transaction: %+v\n", e)
+	}
+
 	ev := e.Event.(*replication.RowsEvent)
 
 	// Caveat: table may be altered at runtime.
@@ -266,9 +270,6 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 		action = DeleteAction
 	case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
 		action = UpdateAction
-	case replication.ANONYMOUS_GTID_EVENT:
-		// Skip anonymous transactions
-		return nil
 	default:
 		return errors.Errorf("%s not supported now", e.Header.EventType)
 	}