thswlsqls opened a new issue, #9054: URL: https://github.com/apache/paimon/issues/9054
**Search before asking** - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. **Paimon version** master @ b15bda5e3 (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?** - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
