fix(mysql): handle binlog incident event (resync required) (#4435)

## Summary

Handle MySQL binlog `INCIDENT_EVENT` (e.g. `LOST_EVENTS`) in the CDC
pull loop. When the source emits an incident event, binlog events were
lost from the stream and our CDC position can no longer be trusted — so
we fail with a resync-required error rather than silently skipping the
gap.

- Detect `INCIDENT_EVENT` (decoded by go-mysql as `GenericEvent`, gated
on the header event type) in `PullRecords` and return a new
`MySQLBinlogIncidentError`.
- Add `parseIncidentEvent` to extract the incident number and message
from the event body.
- Classify `MySQLBinlogIncidentError` as `ErrorNotifyBinlogInvalid` /
`BINLOG_INCIDENT` in the alerting classifier.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: dtunikov

flow/alerting/classifier.go

@@ -1132,6 +1132,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if _, ok := errors.AsType[*exceptions.MySQLBinlogIncidentError](err); ok {
+		return ErrorNotifyBinlogInvalid, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "BINLOG_INCIDENT",
+		}
+	}
+
 	if mysqlGeometryParseError, ok := errors.AsType[*exceptions.MySQLGeometryParseError](err); ok &&
 		strings.Contains(mysqlGeometryParseError.Error(), mysqlGeometryLinearRingNotClosedError) {
 		return ErrorUnsupportedDatatype, ErrorInfo{

flow/alerting/classifier_test.go

@@ -1202,3 +1202,15 @@ func TestUnwrappedGoogleAPIErrorShouldNotBeBigQuery(t *testing.T) {
 		Code:   "UNKNOWN",
 	}, errInfo)
 }
+
+func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewMySQLBinlogIncidentError(1, "LOST_EVENTS")
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err))
+	assert.Equal(t, ErrorNotifyBinlogInvalid, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "BINLOG_INCIDENT",
+	}, errInfo)
+}

flow/connectors/mysql/cdc.go

@@ -4,6 +4,7 @@ import (
 	"cmp"
 	"context"
 	"crypto/tls"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -551,6 +552,14 @@ func (c *MySqlConnector) PullRecords(
 				req.RecordStream.UpdateLatestCheckpointText(updatedOffset)
 				c.logger.Info("rotate", slog.String("name", pos.Name), slog.Uint64("pos", uint64(pos.Pos)))
 			}
+		case *replication.GenericEvent:
+			// INCIDENT_EVENT (LOST_EVENTS) - fail and require resync
+			if event.Header.EventType == replication.INCIDENT_EVENT {
+				incident, message := parseIncidentEvent(ev.Data)
+				c.logger.Error("[mysql] received binlog incident event, resync required",
+					slog.Uint64("incident", uint64(incident)), slog.String("message", message))
+				return exceptions.NewMySQLBinlogIncidentError(incident, message)
+			}
 		case *replication.QueryEvent:
 			if !inTx && gset == nil && event.Header.LogPos > pos.Pos {
 				pos.Pos = event.Header.LogPos
@@ -872,6 +881,20 @@ func posToOffsetText(pos mysql.Position) string {
 	return fmt.Sprintf("!f:%s,%x", pos.Name, pos.Pos)
 }
 
+// parseIncidentEvent extracts the incident number and human-readable message.
+// Best-effort: returns what it can if the body is truncated.
+func parseIncidentEvent(data []byte) (uint16, string) {
+	if len(data) < 2 {
+		return 0, ""
+	}
+	incident := binary.LittleEndian.Uint16(data[:2])
+	if len(data) < 3 {
+		return incident, ""
+	}
+	end := min(3+int(data[2]), len(data))
+	return incident, string(data[3:end])
+}
+
 // processTableMapEventSchema compares the TABLE_MAP_EVENT schema against the cached schema
 // and returns a TableSchemaDelta if new columns are detected (e.g., after gh-ost migration).
 // It also returns a slice mapping binlog column index to FieldDescription for efficient row processing.

flow/shared/exceptions/mysql.go

@@ -137,3 +137,19 @@ func (e *MySQLStaleConnectionError) Error() string {
 	return fmt.Sprintf("MySQL connection is stale: no events received in %v (heartbeat=%v)",
 		e.Since, e.HeartbeatPeriod)
 }
+
+type MySQLBinlogIncidentError struct {
+	Message  string
+	Incident uint16
+}
+
+func NewMySQLBinlogIncidentError(incident uint16, message string) *MySQLBinlogIncidentError {
+	return &MySQLBinlogIncidentError{Incident: incident, Message: message}
+}
+
+func (e *MySQLBinlogIncidentError) Error() string {
+	if e.Message != "" {
+		return fmt.Sprintf("MySQL binlog incident event received (incident=%d): %s; a resync is required", e.Incident, e.Message)
+	}
+	return fmt.Sprintf("MySQL binlog incident event received (incident=%d); a resync is required", e.Incident)
+}