Connect

CDC Schema Registry Patterns

Relational CDC inputs attach each table’s schema to every message they emit. The schema_registry_encode processor reads that schema, registers it in a schema registry, and encodes the message with it, so you can stream schema-encoded change events to Redpanda without registering schemas ahead of time. Use these patterns when downstream consumers expect schema-registry-encoded records from your database tables.

Use this cookbook to:

  • Apply patterns that register CDC table schemas in a schema registry automatically

  • Adapt the patterns to PostgreSQL, MySQL, SQL Server, and Oracle CDC inputs

  • Identify the data types and transformations that prevent a message from encoding

How it works

Each CDC input adds a schema metadata field to the messages it emits. The field holds the source table’s columns and types in a common schema format, and a fingerprint that identifies that version of the table structure.

When you set schema_metadata on the schema_registry_encode processor, the processor uses that metadata instead of fetching a schema from the registry. For each message, the processor:

  1. Reads the schema from the metadata key named in schema_metadata.

  2. Converts the schema to the format set in format.

  3. Registers the schema under the subject set in subject, unless it has already registered that schema version for that subject.

  4. Encodes the message with the schema and prefixes it with the schema ID, in the standard schema registry wire format.

When a table’s structure changes, the fingerprint changes and the processor registers a new schema version under the same subject. The registry’s compatibility rules for that subject decide whether it accepts the new version. If the registry rejects it, the message fails to encode.

The following inputs emit schema metadata:

Input Available from Table name metadata

postgres_cdc

4.82.0

table

mysql_cdc

4.81.0

table

microsoft_sql_server_cdc

4.83.0

table

oracledb_cdc

4.84.0

table_name

The schema_metadata field of the schema_registry_encode processor requires Redpanda Connect 4.83.0 or later.

Prerequisites

Before using these patterns, configure the following.

Redpanda CLI

Install the Redpanda CLI (rpk) to run Redpanda Connect. See rpk installation for installation instructions.

Enterprise license

The CDC inputs in this cookbook require a Redpanda Enterprise Edition license. See Enterprise Licensing.

Source database and schema registry

Configure the source database for CDC as described in the cookbook for your database:

You also need a schema registry that the pipeline can write to, such as Redpanda Schema Registry.

Environment variables

The examples in this cookbook use environment variables for configuration:

export SCHEMA_REGISTRY_URL=http://localhost:8081 (1)
export REDPANDA_BROKERS=localhost:9092 (2)
1 The schema registry URL.
2 The Redpanda broker addresses.

Each example also uses the connection variables from the cookbook for its database, such as PG_DSN, PG_SCHEMA, and PG_SLOT_NAME for PostgreSQL.

Register Avro schemas from PostgreSQL

Encode PostgreSQL change events as Avro and write each table to its own topic:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
      - customers
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - schema_registry_encode:
        url: ${SCHEMA_REGISTRY_URL}
        subject: ${! meta("table") }-value
        schema_metadata: schema
        format: avro
        avro:
          raw_json: true

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("table") }

This pattern:

  • Reads the table schema from the schema metadata field and converts it to an Avro schema.

  • Registers the schema under a subject named after the table, such as orders-value. Because the output writes each table to a topic with the same name, the subject follows the <topic>-value naming convention that most schema registry clients expect.

  • Sets avro.raw_json to true, because CDC inputs emit standard JSON rather than Avro JSON. In schema_metadata mode with the avro format, the processor refuses to start unless you set avro.raw_json explicitly.

Register Avro schemas from MySQL

The same processor configuration works with the mysql_cdc input:

input:
  mysql_cdc:
    dsn: ${MYSQL_DSN}
    tables:
      - orders
      - customers
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - schema_registry_encode:
        url: ${SCHEMA_REGISTRY_URL}
        subject: ${! meta("table") }-value
        schema_metadata: schema
        format: avro
        avro:
          raw_json: true

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("table") }

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}

The mysql_cdc input also needs a checkpoint cache, configured here as a Redis cache resource.

Register Avro schemas from SQL Server

The microsoft_sql_server_cdc input also stores the table name in the table metadata field, so the PostgreSQL processor and output configuration works unchanged. Replace the input:

input:
  microsoft_sql_server_cdc:
    connection_string: ${MSSQL_CONN}
    include:
      - dbo.orders
      - dbo.customers
    stream_snapshot: true

By default, the microsoft_sql_server_cdc input stores its checkpoint in a table in the source database, so it doesn’t need a cache resource.

Register Avro schemas from Oracle

The oracledb_cdc input stores the table name in the table_name metadata field instead of table. Oracle also returns unquoted table names in uppercase, so this pattern lowercases the name before using it for the topic and subject:

input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"

pipeline:
  processors:
    - mapping: |
        meta topic = meta("table_name").lowercase()
    - schema_registry_encode:
        url: ${SCHEMA_REGISTRY_URL}
        subject: ${! meta("topic") }-value
        schema_metadata: schema
        format: avro
        avro:
          raw_json: true

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("topic") }

The mapping processor changes only metadata, so the message body and its schema metadata reach schema_registry_encode unchanged.

Filter events before encoding

Drop events you don’t need and set the message key before encoding:

input:
  postgres_cdc:
    dsn: ${PG_DSN}
    schema: ${PG_SCHEMA}
    tables:
      - orders
    slot_name: ${PG_SLOT_NAME}
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root = if meta("operation") == "delete" { deleted() }
        meta key = this.id.string()
    - schema_registry_encode:
        url: ${SCHEMA_REGISTRY_URL}
        subject: orders-value
        schema_metadata: schema
        format: avro
        avro:
          raw_json: true

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: orders
    key: ${! meta("key") }

This pattern:

  • Drops delete events, passing through snapshot (read), insert, and update events.

  • Copies the primary key into the key metadata field before encoding. After schema_registry_encode runs, the message body is binary Avro, so the output can’t read fields from it with json(). Update this.id to match your table’s primary key column.

  • Uses a fixed subject because the pipeline reads a single table.

Processors that run before schema_registry_encode keep the schema metadata field. Don’t remove metadata before the encoder, for example with meta = deleted(), or messages fail with an error that the schema metadata key wasn’t found.

Consume schema-encoded events

Read the encoded events from Redpanda and decode them back to JSON with the schema_registry_decode processor:

input:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topics:
      - orders
    consumer_group: orders-reader

pipeline:
  processors:
    - schema_registry_decode:
        url: ${SCHEMA_REGISTRY_URL}
        avro:
          raw_unions: true

output:
  stdout:
    codec: lines

The decoder reads the schema ID from each message and fetches the matching schema from the registry, so it doesn’t need a subject. The generated Avro schemas make every column nullable, so each value is an Avro union. Setting avro.raw_unions to true decodes values as plain JSON, such as "id":1, instead of Avro JSON, such as "id":{"int":1}.

Troubleshoot encoding errors

When a message fails to encode, schema_registry_encode leaves it unchanged and flags it with an error. If you don’t handle the error, the output writes the unencoded JSON message to the topic. Handle these messages with the error handling processors, for example by routing them to a dead-letter topic.

Message body doesn’t match the table schema

The registered schema describes the source table’s columns. If a processor reshapes the message body before schema_registry_encode, the body no longer matches the schema:

  • Avro encoding drops fields that aren’t in the schema. Versions earlier than 4.84.1 fail to encode the message instead.

  • Avro encoding writes null for a column that’s missing from the body, because the generated schemas make every column optional. The message encodes without an error, so a column that a processor drops by mistake reaches the topic as null.

To publish a different shape, transform the message after it’s decoded downstream, or use schema_registry_encode without schema_metadata and a schema you register yourself.

JSON columns

The generated Avro schema types PostgreSQL json and jsonb columns, MySQL JSON columns, and PostgreSQL column types that have no schema mapping as bytes. The CDC inputs decode these columns into JSON objects, so the avro format fails to encode any row where the column holds a value:

avro: field attrs: cannot use map[string]any with Avro type bytes

Rows where the column is null encode without an error.

To encode these rows, convert the column to a JSON string with a mutation processor before schema_registry_encode. This example converts a column named attrs:

- mutation: |
    root.attrs = if this.attrs != null { this.attrs.format_json(no_indent: true).string() }

Use mutation rather than mapping, so the rest of the message body and its metadata stay unchanged. The .string() call matters: format_json() returns bytes, which the message body would carry as a base64 string. Downstream consumers receive the column as a JSON string and parse it after decoding.

Unconstrained numeric columns

Avro decimals need a fixed precision and scale. The following column types have neither, so the avro format fails to convert schemas that contain them:

  • PostgreSQL NUMERIC declared without precision and scale.

  • Oracle NUMBER declared without precision, FLOAT, and NUMBER columns whose scale is greater than the precision or negative.

Declare the column with a precision and scale, such as NUMERIC(10,2) or NUMBER(10,2).

Transaction markers

When you enable include_transaction_markers on the postgres_cdc input, begin and commit messages carry no schema metadata, so they fail to encode. Drop them before schema_registry_encode:

- mapping: |
    root = if meta("operation") == "begin" || meta("operation") == "commit" { deleted() }

JSON Schema format

The format field also accepts json_schema. In this format, the processor registers a JSON Schema and validates each message against it instead of encoding to binary. The generated JSON Schema treats nullable columns as optional properties but doesn’t accept null as a value, so a row that contains a NULL column fails validation. Use the avro format for tables with nullable columns.

Next steps

Suggested labs

Search all labs