feat(source): support kafka-connect schema guided debezium parsing

[fuyufjh]

When consuming from debezium, the unit info is actually available as part of the schema:

• If using avro, it is in the connect.name extension field. Right now we are ignoring this field, and there is an existing issue for it Add support for Avro schemas using some common Kafka Connect schema names/types #14703
• If using json, there is also schema per message. It would appear as name of a type (not to be confused as name of a field). See the example below. We are also ignoring this name field in source right now. And only added partial support for it in the sink part recently kafka-cdc-sink: Missing name field in debezium json #13450
• Having schema per json message is costly. So debezium does recommend avro or allow disabling schema in json. In the latter case unit info would be unavailable, and we would end up with the same problem here. There is also an existing sink issue requesting the ability to disable. sink message of debezium json without schema field #11539
  • Not debezium but our upsert json sink does not include schema by default but it can be turned on.

We will focus on the second case here. Below is a message from debezium 2.4 MySQL connector:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "string",
            "optional": true,
            "name": "io.debezium.data.Enum",
            "version": 1,
            "parameters": {
              "allowed": "true,last,false,incremental"
            },
            "default": "false",
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "sequence"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "total_order"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "data_collection_order"
          }
        ],
        "optional": true,
        "name": "event.block",
        "version": 1,
        "field": "transaction"
      }
    ],
    "optional": false,
    "name": "dbserver1.inventory.bar.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "bid": 8,
      "at": 1712132700123457
    },
    "source": {
      "version": "2.4.2.Final",
      "connector": "mysql",
      "name": "dbserver1",
      "ts_ms": 1712132719000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": null,
      "table": "bar",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 3181,
      "row": 0,
      "thread": 8,
      "query": null
    },
    "op": "c",
    "ts_ms": 1712132719617,
    "transaction": null
  }
}

The relevant part is here:

      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": true,
            "field": "bid"
          },
          {
            "type": "int64",
            "optional": true,
            "name": "io.debezium.time.MicroTimestamp",
            "version": 1,
            "field": "at"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.bar.Value",
        "field": "after"
      },

and

    "after": {
      "bid": 8,
      "at": 1712132700123457
    },

Note the field after.at has "type": "int64" and "name": "io.debezium.time.MicroTimestamp"

The structure of this schema field is actually from Java class org.apache.kafka.connect.json.JsonConverter:
https://github.com/apache/kafka/blob/3.7.0/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java#L313

Originally posted by @xiangjinwu in #16097 (comment)