- fix logs

- add AdditionalAttributes to classified error

Author: dtunikov

flow/alerting/classifier.go

@@ -1157,10 +1157,17 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
-	if _, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok {
+	if partialRowEventError, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok {
+		var additionalAttributes map[AdditionalErrorAttributeKey]string
+		if partialRowEventError.Schema != "" || partialRowEventError.Table != "" {
+			additionalAttributes = map[AdditionalErrorAttributeKey]string{
+				ErrorAttributeKeyTable: fmt.Sprintf("%s.%s", partialRowEventError.Schema, partialRowEventError.Table),
+			}
+		}
 		return ErrorNotifyBinlogPartialRowEventUnsupported, ErrorInfo{
-			Source: ErrorSourceMySQL,
-			Code:   "UNSUPPORTED_PARTIAL_ROW_EVENT",
+			Source:               ErrorSourceMySQL,
+			Code:                 "UNSUPPORTED_PARTIAL_ROW_EVENT",
+			AdditionalAttributes: additionalAttributes,
 		}
 	}
 

flow/alerting/classifier_test.go

@@ -1253,5 +1253,8 @@ func TestMySQLUnsupportedPartialRowEventShouldBeNotifyPartialRowEventUnsupported
 	assert.Equal(t, ErrorInfo{
 		Source: ErrorSourceMySQL,
 		Code:   "UNSUPPORTED_PARTIAL_ROW_EVENT",
+		AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+			ErrorAttributeKeyTable: "e2e_test.partial_rows",
+		},
 	}, errInfo)
 }

flow/connectors/mysql/cdc.go

@@ -583,7 +583,7 @@ func (c *MySqlConnector) PullRecords(
 			attribute.String(otel_metrics.BinlogEventTypeKey, fmt.Sprintf("%s_%d", prefix, int(event.Header.EventType))),
 		)))
 		if _, loaded := c.warnedUnsupportedEventTypes.LoadOrStore(event.Header.EventType, struct{}{}); !loaded {
-			c.logger.Warn("unsupported rows event", slog.Any("type", event.Header.EventType))
+			c.logger.Warn("unsupported binlog event", slog.Any("type", event.Header.EventType))
 		}
 	}
 
@@ -634,6 +634,10 @@ func (c *MySqlConnector) PullRecords(
 					break
 				}
 				tableID, totalFragments, ok := parsePartialRowEventTableID(ev.Data)
+				// tableIdToName tracking relies on TABLE_MAP_EVENT and
+				// PARTIAL_ROW_DATA_EVENT coming within the same batch, so no checkpoints in between
+				// Server sends events in groups with the usual sequence of begin - table map - rows - gtid/commit
+				// So as of committing this, the unresolvable path is not expected to be hit
 				sourceTableName, known := tableIdToName[tableID]
 				if !ok || !known {
 					// couldn't recover/resolve the table, fail loudly

flow/otel_metrics/otel_manager.go

@@ -607,7 +607,7 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error {
 
 	if om.Metrics.UnsupportedBinlogEventCounter, err = om.GetOrInitInt64Counter(BuildMetricName(UnsupportedBinlogEventName),
 		metric.WithDescription(
-			"Counter of unsupported binlog generic events seen on the CDC path, with `eventType` label "+
+			"Counter of unsupported binlog events seen on the CDC path, with `eventType` label "+
 				"holding the numeric binlog event type"),
 	); err != nil {
 		return err