[Bug] [Source][MySQL-CDC] compatible debezium json format npe issue

[liunaijie]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

mysql cdc when use default format, the update will generate 2 record (1 delete, 1 update)
when use compatible_debezium_json format, the update will only generate 1 record
But when set key_convert_schemas.enable to false, will get NPE

SeaTunnel Version

dev

SeaTunnel Config

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

source {
  MySQL-CDC {
    base-url = "jdbc:mysql://localhost:3306/st_cdc"
    username = "root"
    password = "123456"
    table-names = ["st_cdc.tb01"]
    startup.mode = "initial"
    format = compatible_debezium_json
    debezium = {
        key.converter.schemas.enable = false
        value.converter.schemas.enable = false
    }
  }
}

sink {
  Console {
  }
}

### Running Command

```shell
example

Error Exception

Null point exception

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

when set to true, the body will be empty.

when set to false, get npe

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[liunaijie]

another thing i noticed, when i use mysql cdc to mysql cdc with default format, the update will insert new record, but can't delete the old record.

when i change to compatible_debezium_json format, the data will be debezium format, can't sink to mysql directly.