[FLINK-36611][pipeline-connector][kafka] Add schema info to output of Kafka sink

[MOBIN-F]

Currently, the output of Kafka sink in debezium format looks like this:

{
  "before": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "2222",
    "age": 12
  },
  "after": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "1234",
    "age": 12
  },
  "op": "u",
  "source": {
    "db": null,
    "table": "customers"
  }
} 

It contains record data with full before/after and db info, but schema info wasn't included.

However, In some scenarios, we need this information to determine the type of data. For example, Paimon's Kafka CDC source requires this type information, otherwise all types are considered String, refer to https://paimon.apache.org/docs/0.9/flink/cdc-ingestion/kafka-cdc/#supported-formats.

Considering that this will increase the data load, I suggest adding a parameter to configure whether to enable it.

[MOBIN-F]

Because the schema is passed to the downstream as a string, and there is a nested json in the schema, if the json string is put into jsonNode, there will be ["]. The JsonNode.asText() method can solve this problem well.

[lvyanquan]

Hi, @MOBIN-F. Is there any blocker for this being ready?

[lvyanquan]

This includes ARRAY/MAP/ROW, There may be some issues if all they were converted to SchemaBuilder.string.
You can check if there is a better type, and if not, it is also acceptable.

[MOBIN-F]

debezium-json does not seem to receive ARRAY/MAP/ROW type data because the mysql does not support this type

[lvyanquan]

Thanks @MOBIN-F for this update.
Overall, This modification is good for me, left some comments about the code structure.

Additionally, you can manually test the data types in Paimon action to verify its correctness.

[MOBIN-F]

Do we need to add Literal/Semantic type to the documentation? It seems like users won't care about it? @lvyanquan

[lvyanquan]

Yes, I suggest not add this as users won't pay much attention on it.

I think adding more information is not a disadvantage, so we can keep them.

[MOBIN-F]

Thanks @MOBIN-F for this update. Overall, This modification is good for me, left some comments about the code structure.

Additionally, you can manually test the data types in Paimon action to verify its correctness.

I tested 【pipeline-kafka-->kafka-->paimon action->paimon table】, and it worked as expected. Maybe I can add some e2e tests later.

[lvyanquan]

LGTM.

Left some minor comments.

[lvyanquan]

LGTM.

[lvyanquan]

Hi @leonardBang @ruanhang1993 could you help to check about this?

[leonardBang]

Thanks @MOBIN-F and @lvyanquan for the contribution, LGTM, wait the CI green

[MOBIN-F]

CI failed, but it has nothing to do with this PR because this PR only involves the kafka-pipeline module