Debezium JSON

The debezium_json codec decodes Debezium CDC (Change Data Capture) Envelope JSON into a columnar Arrow MessageBatch. Attach it to a Kafka input that consumes a topic written by Debezium to turn database change events (c/u/d/r) into queryable rows. CDC offset is not managed here — it is the Kafka input's ack-gated offset.

Configuration

Field	Type	Required	Default	Description
type	string	yes	—	Fixed value "debezium_json"

This codec takes no configuration fields; build ignores config. It is only attached as a codec hook on a Kafka input.

Examples

yaml

input:
  type: kafka
  brokers:
    - localhost:9092
  topics:
    - shop.users
  consumer_group: arkflow-cdc
  start_from_latest: false
  codec:
    type: debezium_json

Deployment topology:

text

Database → Debezium (Kafka Connect / Debezium Server) → Kafka topic → ArkFlow (kafka input + debezium_json codec)

See examples/cdc_debezium.yaml.

Semantics

Output schema

Each Debezium Envelope { before, after, op, source, ts_ms } is flattened into one row:

Column	Source	Notes
<business fields>	after (falls back to before on deletes)	Promoted to top-level columns
op	op	c / u / d / r
ts_ms	ts_ms	Change timestamp
source_db, source_table	source.db, source.table	Top-level scalar columns
before	Full before object	JSON text column (parse with SQL JSON functions)
source	Full source object	JSON text column

before / source are kept as JSON text (rather than nested structs) because mixing nulls and objects within the same batch (e.g. an insert's before is null while an update's is an object) would conflict with the Arrow JSON reader's single-pass schema inference.

Delivery semantics

• The CDC offset is provided by the Kafka input's ack-gated offset (at-least-once). Ensure downstream sinks are idempotent.
• When op="d", after is null and business fields are taken from before.

Name collisions with business fields

If the business payload itself contains a field named op, ts_ms, source_db, source_table, before or source, the business value keeps its column and the envelope metadata value moves to a reserved __debezium_<name> column, with one warning per batch. Without a collision the metadata columns keep their canonical names, so existing pipelines are unaffected.

Tombstones

A zero-length payload (a Debezium tombstone — Kafka null-value messages are usually intercepted by the Kafka input before the codec) is skipped with a warning instead of failing the batch; the remaining messages decode normally. A JSON literal null is not a tombstone and keeps its existing all-null-row behavior.

Notes / Non-goals

• Does not connect directly to MySQL binlog / PostgreSQL logical replication (planned as a separate input in the future).
• Only Debezium JSON is supported; Avro / Protobuf formats are not supported yet (Avro requires a Schema Registry).