fix(mysql): classify event checksum error (#4537)

verify event checksum and throw binlon invalid error if invalid

Author: dtunikov

flow/alerting/classifier.go

@@ -17,6 +17,7 @@ import (
 	chproto "github.com/ClickHouse/ch-go/proto"
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5/pgconn"
 	"go.mongodb.org/mongo-driver/v2/mongo"
@@ -1129,6 +1130,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if strings.Contains(err.Error(), replication.ErrChecksumMismatch.Error()) {
+		return ErrorNotifyBinlogInvalid, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "BINLOG_CHECKSUM_MISMATCH",
+		}
+	}
+
 	if mysqlExecuteError, ok := errors.AsType[*exceptions.MySQLExecuteError](err); ok {
 		errClass := ErrorOther
 		if mysqlExecuteError.Retryable {

flow/alerting/classifier_test.go

@@ -15,6 +15,7 @@ import (
 	chproto "github.com/ClickHouse/ch-go/proto"
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -907,6 +908,17 @@ func TestMySQLBinlogEventExceededMaxAllowedPacket(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestMySQLBinlogChecksumMismatch(t *testing.T) {
+	err := exceptions.NewMySQLExecuteError(
+		fmt.Errorf("failed checksum for WriteRowsEventV2, log pos 12345: %v", replication.ErrChecksumMismatch))
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed in pull records: %w", err))
+	assert.Equal(t, ErrorNotifyBinlogInvalid, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "BINLOG_CHECKSUM_MISMATCH",
+	}, errInfo, "Unexpected error info")
+}
+
 func TestMySQLExecuteError(t *testing.T) {
 	err := exceptions.NewMySQLExecuteError(
 		tls.RecordHeaderError{Msg: "first record does not look like a TLS handshake"})

flow/connectors/mysql/cdc.go

@@ -350,6 +350,7 @@ func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string)
 		DisableRetrySync: true,
 		UseDecimal:       true,
 		ParseTime:        true,
+		VerifyChecksum:   true,
 		TLSConfig:        tlsConfig,
 		HeartbeatPeriod:  c.binlogHeartbeatPeriod,
 		EventCacheCount:  eventCacheCount,