[paimon-flink-cdc] Add skip-corrupt-record configuration for Kafka CDC with PII-safe logging

[0dunay0]

Purpose

Fixes job failures when Kafka CDC ingestion encounters corrupt or unparsable records. Instead of crashing the job, users can now configure the parser to skip invalid records with PII-safe logging.

This feature is already available for MySQL sources.

Tests

It's a difficult thing to test to produce invalid avro messages to a topic but it's been tested with a live job in our cluster.

2025-10-01 00:04:51,433 [] WARN org.apache.paimon.flink.action.cdc.format.AbstractRecordParser [] - Skipping corrupt or unparsable source record.

Documentation

Docs has been updated.