[Bug] NPE when metadata columns are used with debezium-bson format

[thswlsqls]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version
master @ b15bda5 (2.1-SNAPSHOT)

Compute Engine
Flink (paimon-flink-cdc)

Minimal reproduce step

1. Run kafka_sync_table (or kafka_sync_database) with --kafka_conf value.format=debezium-bson.
2. Add any --metadata_column topic (or offset, partition, timestamp, timestamp_type).
3. The target table is created with the metadata columns, then the job fails on the first record.

What doesn't meet your expectations?
Expected: metadata columns are filled with the Kafka message metadata, as documented in docs/docs/cdc-ingestion/kafka-cdc.mdx.
Actual: NullPointerException in KafkaMetadataConverter.read(), so the job restarts in a loop and no data is ingested.

Anything else?
DebeziumBsonRecordParser.setRoot() overrides AbstractRecordParser.setRoot() without storing the record, so AbstractRecordParser.evalMetadataColumns() passes a null currentRecord to every converter. The sibling parsers AbstractJsonRecordParser, DebeziumJsonRecordParser and DebeziumAvroRecordParser all store it; the metadata column feature (PR #7315) added the consumer to the BSON parser but not the assignment.

Are you willing to submit a PR?

• I'm willing to submit a PR!