9. Transformations

Single Message Transforms are Kafka Connect’s per-record hook. They are useful, limited by design, and routinely over-used.

What an SMT is

   a function: one record in, one record out
   (or null, to drop it)

   runs in the connector's own thread
   BEFORE the converter

   ✓ rename, reshape, mask, route, filter,
     add fields
   ✗ NO STATE — it cannot remember the
     previous record
   ✗ NO JOINS — it cannot look anything up
   ✗ NO AGGREGATION
   ✗ NO REORDERING

Stateless and per-record. Every limitation follows from that, and so does the rule for when to stop.

transforms=unwrap,mask,route
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.mask.type=org.apache.kafka.connect.transforms.MaskField$Value
transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter

They run in the listed order, left to right, and order matters a great deal — an SMT that expects a flat row must come after the unwrap.

ExtractNewRecordState

The one nearly every deployment uses.

transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields=op,source.ts_ms,source.lsn,source.table
transforms.unwrap.add.headers=op
transforms.unwrap.delete.tombstone.handling.mode=rewrite

   BEFORE                      AFTER
   {before:{...},              {id: 4471,
    after:{id:4471,             status: "shipped",
           status:"shipped"},   amount: "149.99",
    source:{...},               __op: "u",
    op:"u", ts_ms:...}          __source_ts_ms: ...,
                                __source_lsn: ...,
                                __deleted: "false"}

It throws away the envelope and keeps after. Most sink connectors expect a flat row, so this is close to mandatory.

Delete handling

The part that needs a decision.

   drop      delete events are DISCARDED
             ⚠ the sink never learns the row
               is gone. it stays forever.

   rewrite   the `before` row is emitted with
             __deleted = "true"
             ✓ the sink can act on it
             ← usually what you want

   none      the delete becomes a null value
             ✓ correct for compaction and
               for sinks that treat null as
               delete

drop is the default in some configurations and it silently accumulates deleted rows in your sink. If your search index has orders that no longer exist, this is why.

Property names here have changed across versions — older delete.handling.mode versus newer delete.tombstone.handling.mode. Check yours; a silently-ignored misspelled property gives you the default.

What you lose

   after unwrapping, you no longer have:
     the `before` image
     the full source block
     the operation, unless you add it

add.fields=op,source.ts_ms is worth setting by default. A consumer that cannot tell an insert from an update, or does not know the event time, is missing information it will want later — and adding it retroactively means re-snapshotting.

Use add.headers for anything routing or filtering should see without deserialising the value.

Filtering

transforms=filter
transforms.filter.type=io.debezium.transforms.Filter
transforms.filter.language=jsr223.groovy
transforms.filter.condition=\
  value.op != 'd' && value.after.region == 'EMEA'

Drops records the expression rejects.

   ⚠ THREE WARNINGS

   1  A SCRIPTING ENGINE, PER RECORD.
      real CPU cost at volume.

   2  FILTER AT THE SOURCE INSTEAD, WHERE
      POSSIBLE.
      table.include.list and column.include.list
      never read the data at all — vastly
      cheaper than reading and discarding.

   3  A DROPPED RECORD IS GONE.
      change the filter later and the data
      you skipped is not in the topic.
      → an incremental snapshot (chapter 5)
        is the recovery, and only if the
        rows still exist.

Prefer source-side filtering:

table.include.list=public.orders,public.customers
column.exclude.list=public.customers.password_hash,public.customers.ssn

column.exclude.list is the better tool for excluding sensitive columns, because the value never enters the pipeline at all — nothing to leak from a topic, a log or a dead-letter queue.

Masking

transforms=mask
transforms.mask.type=org.apache.kafka.connect.transforms.MaskField$Value
transforms.mask.fields=email,phone
transforms.mask.replacement=***

Debezium’s own column-level masking and hashing happen earlier, at the connector:

column.mask.with.12.chars=public.customers.card_number
column.mask.hash.SHA-256.with.salt.mySalt=public.customers.email

   HASHING is the useful one:
     the same input gives the same hash
     → you can still JOIN and GROUP BY it
     → without holding the value

Hashing preserves referential utility while removing the data, which is often exactly the requirement. Use the connector’s own masking rather than an SMT where you can — it applies before the value is ever in a record.

Other useful transforms

   TimezoneConverter      normalise temporal
                          columns to a target
                          zone
   HeaderToValue /
   ValueToHeader          move fields between
                          the record and its
                          headers
   ExtractChangedRecordState
                          add headers naming
                          which fields changed
   EventRouter            the OUTBOX router —
                          chapter 10
   InsertField            add static metadata
   ReplaceField           rename or drop fields
   Cast                   change a field's type

ExtractChangedRecordState is quietly useful: it annotates each record with which columns actually changed, so a consumer can skip work when the change was irrelevant to it.

Predicates

Apply an SMT conditionally.

transforms=dropDeletes
transforms.dropDeletes.type=org.apache.kafka.connect.transforms.Filter
transforms.dropDeletes.predicate=isAudit
transforms.dropDeletes.negate=false

predicates=isAudit
predicates.isAudit.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.isAudit.pattern=cdc\\.public\\.audit_.*

Different treatment per topic within one connector, without splitting into several connectors. Genuinely useful when one connector covers tables that need different handling.

Where to stop

The important part of the chapter.

   ✓ AN SMT IS RIGHT FOR
     unwrapping the envelope
     renaming a topic or a field
     masking or hashing a column
     adding metadata
     dropping a record on a simple condition
     routing sharded tables together

   ✗ USE A STREAM PROCESSOR FOR
     joining two tables
     aggregating
     deduplicating across records
     anything needing the PREVIOUS record
     enrichment from an external system
     windowing, or anything with time
     complex conditional business logic

The signal is simple: if the transform needs to remember anything, it is not an SMT.

   Debezium  ──▶  Kafka  ──▶  FLINK  ──▶  sinks
                            ▲
                            joins, aggregates,
                            deduplication,
                            enrichment, windows

Flink’s chapter 10 covers changelog streams directly, and its temporal joins chapter describes joining a CDC stream against a versioned table — “the join that makes CDC worth it”, as it puts it. That is the tool for anything past the list above.

Three reasons to keep the connector thin, beyond capability:

Throughput. SMTs run in the single connector task (chapter 3), which cannot be scaled out. Every millisecond per record is a millisecond off your ceiling. A scripting-engine SMT at 20,000 records a second is a real bottleneck.

Debuggability. A misbehaving SMT chain is diagnosed by reading connector configuration and restarting the pipeline. A misbehaving Flink job has a UI, tests and a deployment story.

Blast radius. A bad SMT stops the connector, which stops the pipeline, which — on PostgreSQL — starts filling the source database’s disk (chapter 2). A bad Flink job stops a Flink job.

That last reason is the strongest one. Anything you put in the connector is on the critical path of your source database’s health.

Error handling

errors.tolerance=all
errors.log.enable=true
errors.log.include.messages=true
errors.deadletterqueue.topic.name=cdc-dlq
errors.deadletterqueue.context.headers.enable=true
errors.deadletterqueue.topic.replication.factor=3

   errors.tolerance=none    (default)
     one bad record STOPS THE CONNECTOR
     → and on Postgres, the WAL starts
       accumulating

   errors.tolerance=all
     bad records go to the DLQ, the
     connector continues
     ⚠ AND YOU HAVE SILENTLY LOST DATA
       unless someone watches the DLQ

Neither default is safe on its own. none turns a single malformed record into a database incident; all turns it into invisible data loss.

The workable answer is all, plus an alert on DLQ depth, plus a runbook. Set context.headers.enable=true so DLQ records carry the original topic, partition, offset and the exception — without it, a DLQ record is a payload with no explanation.

Note that this covers converter and SMT errors, not connector failures. A connector that cannot read the log fails regardless of tolerance.

What to take from this chapter

• An SMT is one record in, one record out, stateless, before the converter. Every limitation follows from that.
• They run in the order listed, and order matters — anything expecting a flat row goes after the unwrap.
• ExtractNewRecordState is near-mandatory, and its delete handling needs a deliberate choice — drop silently leaves deleted rows in your sink forever.
• Add op and source.ts_ms back with add.fields, or consumers lose information that cannot be recovered without a re-snapshot.
• Filter at the source with table.include.list and column.exclude.list, not with a scripting SMT — the data is never read at all, and excluded columns never enter the pipeline.
• Hashing preserves joinability while removing the value, and connector-level masking beats an SMT because it applies earlier.
• Predicates apply an SMT to selected topics only.
• If a transform needs to remember anything, it is not an SMT — it is a Flink job.
• Keep the connector thin for three reasons: it is single-task and cannot scale, it is hard to debug, and anything that breaks it starts filling your source database’s disk.
• Set errors.tolerance=all with a DLQ, context headers, and an alert. none causes database incidents; all alone causes silent loss.

Next: the outbox pattern — how to publish events you designed, instead of the table rows you happen to have.