[FLINK-32743][Connectors/Kafka] Parse data from kafka connect and convert it into regular JSON data

[sunxiaojian]

Add "record.key.include.kafka.connect.json.schema" and "record.value.include.kafka.connect.json.schema" configurations. When this configuration is true, records can be converted into regular JSON data

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[MartijnVisser]

@sunxiaojian Can you please rebase your PR?

[sunxiaojian]

@sunxiaojian Can you please rebase your PR?

@MartijnVisser Thanks for your review, it has been processed

[MartijnVisser]

Thanks for your review, it has been processed

CI is failing, will you take a look?

[sunxiaojian]

Thanks for your review, it has been processed

CI is failing, will you take a look?

@MartijnVisser fixed

[MartijnVisser]

@mas-chen Want to take a look?

[mas-chen]

@sunxiaojian have you considered implementing this as a Flink format? You could draw inspiration from pre-existing formats: https://github.com/apache/flink/tree/master/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json

For example, the debezium json format already has an integration with Kafka connect.

[sunxiaojian]

@sunxiaojian have you considered implementing this as a Flink format? You could draw inspiration from pre-existing formats: https://github.com/apache/flink/tree/master/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json

For example, the debezium json format already has an integration with Kafka connect.

@mas-chen ok, that's a good suggestion