don't warn on the same event type twice

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -579,10 +579,12 @@ func (c *MySqlConnector) PullRecords(
 		}
 	}
 	recordUnsupportedEvent := func(ctx context.Context, event *replication.BinlogEvent, prefix string) {
-		c.logger.Warn("unsupported rows event", slog.Any("type", event.Header.EventType))
 		otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
 			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))
+		}
 	}
 
 	lastEventAt := time.Now()

flow/connectors/mysql/mysql.go

@@ -31,18 +31,19 @@ import (
 
 type MySqlConnector struct {
 	*metadataStore.PostgresMetadata
-	config                *protos.MySqlConfig
-	ssh                   *utils.SSHTunnel
-	conn                  atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe
-	contexts              atomic.Pointer[chan context.Context]
-	logger                log.Logger
-	rdsAuth               *utils.RDSAuth
-	serverVersion         string
-	collationCharset      atomic.Pointer[map[uint64]string]
-	warnedCharsets        sync.Map
-	binlogHeartbeatPeriod time.Duration
-	totalBytesRead        atomic.Int64
-	deltaBytesRead        atomic.Int64
+	config                      *protos.MySqlConfig
+	ssh                         *utils.SSHTunnel
+	conn                        atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe
+	contexts                    atomic.Pointer[chan context.Context]
+	logger                      log.Logger
+	rdsAuth                     *utils.RDSAuth
+	serverVersion               string
+	collationCharset            atomic.Pointer[map[uint64]string]
+	warnedUnsupportedEventTypes sync.Map
+	warnedCharsets              sync.Map
+	binlogHeartbeatPeriod       time.Duration
+	totalBytesRead              atomic.Int64
+	deltaBytesRead              atomic.Int64
 }
 
 func NewMySqlConnector(ctx context.Context, config *protos.MySqlConfig) (*MySqlConnector, error) {