Search before asking

• I searched in the issues and found nothing similar.

Motivation

cdc action supports the following formats of kafka data which collected from mongodb via debezium, regardless of whether it contains schema

{
  "before": null,
  "after": "{\"_id\":{\"$oid\":\"64001c996f4de7ff3189d374\"},\"last_updated_at\":{\"$numberLong\":\"1732232838425\"},\"tags\":[\"completely\",\"pass\"],\"updated_by\":\"xxx\"}",
  "updateDescription": null,
  "source": {
    "version": "2.7.0.Final",
    "connector": "mongodb",
    "name": "datapipeline",
    "ts_ms": 1732644484000,
    "snapshot": "false",
    "db": "datapipeline",
    "sequence": null,
    "ts_us": 1732644484000000,
    "ts_ns": 1732644484000000000,
    "collection": "clips",
    "ord": 22,
    "lsid": null,
    "txnNumber": null,
    "wallTime": null
  },
  "op": "c",
  "ts_ms": 1732644484231,
  "transaction": null
}

Solution

Add debezium-bson format for kafka cdc action:

1. support parse bson string from before/after field in kafka message
2. extract java object and basic data type from bson value
3. convert java basic data type to string, java object to json string
4. supports schema evolution, and all fields are fixed to string type, _id is fixed to primary key

Expected Results：

• Schema:

• Records:

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!