debezium.encode

Encodes records into the messages Debezium would have produced.

Not usable in v0.21

This processor is in the Conduit repository but is not registered as a built-in processor in v0.21, the Postgres source has no Debezium output mode, and the Kafka destination cannot write the messages this processor produces. The configuration below documents the design, not a working pipeline. Debezium-compatible output for Postgres did not ship in v0.21 and is not scheduled.

Description

The debezium.encode processor is the envelope, naming, and tombstone half of Conduit's Debezium-compatible output mode for Postgres CDC. It takes each OpenCDC record the Postgres source emits in its Debezium mode and turns it into the messages Debezium would have produced:

• an insert or snapshot read becomes one create/read message;
• an update becomes one update message, or, when the key changes, a delete, a tombstone and a create;
• a delete becomes a delete message and, with tombstones.on.delete, a tombstone (same key, null value).

Each message carries its serialized key, value and headers, and the ordered list is written as a versioned frame into the payload. The Kafka destination produces every message of the frame and acknowledges the record only when all of them succeeded, so positions and change keys stay one-record-per-change.

This revision implements the JsonConverter (schemas.enable true and false, decimal.format BASE64 and NUMERIC). The Confluent AvroConverter is the design's default and the next revision; selecting avro fails at configure time with debezium.avro_unsupported rather than emitting half-encoded messages.

The processor works together with the Postgres source's Debezium mode. The debezium.compat setting must match the source's setting, and migrate sets both identically when it emits a Debezium Postgres pipeline.

Configuration parameters

YAML

version: 2.2
pipelines:
  - id: example
    status: running
    connectors:
      # define source and destination ...
    processors:
      - id: example
        plugin: "debezium.encode"
        settings:
          # Selects the key converter. The Debezium-compatible output design
          # makes "avro" the default; this revision implements the JsonConverter
          # first and defaults to "json" until the Confluent AvroConverter port
          # lands.
          # Type: string
          key.converter: "json"
          # Selects the value converter. See key.converter.
          # Type: string
          value.converter: "json"
          # Mirrors key.converter.schemas.enable: when true the JsonConverter
          # includes the schema next to the key payload.
          # Type: bool
          key.json.schemas.enable: "true"
          # Mirrors value.converter.schemas.enable: when true the JsonConverter
          # includes the schema next to the value payload.
          # Type: bool
          value.json.schemas.enable: "true"
          # Mirrors decimal.format: "BASE64" writes the unscaled decimal as
          # base64, "NUMERIC" as a JSON number.
          # Type: string
          json.decimal.format: "BASE64"
          # Mirrors tombstones.on.delete: a delete expands into a delete event
          # followed by a tombstone (same key, null value).
          # Type: bool
          tombstones.on.delete: "true"
          # Selects the Debezium source-schema shape. It must match the
          # source's debezium.compat.
          # Type: string
          debezium.compat: "3.5"

Table

Name	Type	Default	Description
key.converter	string	json	Selects the key converter. The Debezium-compatible output design makes avro the default; this revision implements the JsonConverter first and defaults to json until the Confluent AvroConverter port lands.
value.converter	string	json	Selects the value converter. See key.converter.
key.json.schemas.enable	bool	true	Mirrors key.converter.schemas.enable: when true the JsonConverter includes the schema next to the key payload.
value.json.schemas.enable	bool	true	Mirrors value.converter.schemas.enable: when true the JsonConverter includes the schema next to the value payload.
json.decimal.format	string	BASE64	Mirrors decimal.format: BASE64 writes the unscaled decimal as base64, NUMERIC as a JSON number.
tombstones.on.delete	bool	true	Mirrors tombstones.on.delete: a delete expands into a delete event followed by a tombstone (same key, null value).
debezium.compat	string	3.5	Selects the Debezium source-schema shape. It must match the source's debezium.compat.

Examples

The processor is used as part of a Debezium Postgres → Kafka pipeline, where the Postgres source runs in Debezium mode and debezium.encode sits on the source connector before the Kafka destination:

version: 2.2
pipelines:
  - id: pg-debezium-to-kafka
    status: running
    connectors:
      - id: pg
        type: source
        plugin: builtin:postgres
        settings:
          url: "postgres://user:pass@localhost:5432/db?sslmode=disable"
          tables: "inventory.orders"
          debezium.enabled: "true"
        processors:
          - id: encode
            plugin: debezium.encode
            settings:
              debezium.compat: "3.5"
      - id: kafka
        type: destination
        plugin: builtin:kafka
        settings:
          servers: "localhost:9092"
          topic: '{{ index .Metadata "debezium.topic" }}'