Kafka Connect ExtractNewRecordState (Debezium) SMT Usage Reference for Confluent Cloud

The ExtractNewRecordState SMT (io.debezium.transforms.ExtractNewRecordState) flattens Debezium change events by replacing the event envelope with the row state after the change.

Description

Debezium change events use a nested envelope with before, after, source, op, and ts_ms fields. Use this SMT when downstream consumers or sink connectors require a flat record instead of the envelope.

In addition to unwrapping the after state, you can configure this SMT to:

Control how delete events and tombstones are emitted.
Copy change event metadata into the record value or headers.
Replace null column values with their column default values.

For complete details, see the official Debezium New Record State Extraction documentation. For the full list of available SMTs, see Kafka Connect Single Message Transformation Reference for Confluent Cloud.

Note

When using this SMT, consider the following:

Place this SMT after SMTs that read the change event envelope, such as ExtractChangedRecordState, and before SMTs that operate on the flattened row fields. The record key and headers set by earlier SMTs, such as ToLogicalTopicRouter, are preserved.
Don’t use unwrap as the transform alias for the managed Debezium change data capture (CDC) Source connectors. The connector reserves this alias for after.state.only. For details, see Transform alias validation.
If you set after.state.only to true, the connector already flattens records, so the options of this SMT have no effect. Set after.state.only to false to configure this SMT.
replace.null.with.default applies only to columns that have a default value defined in the source database. Null values in columns without a default remain null.

Limitations

The ExtractNewRecordState SMT is available only for the Oracle XStream CDC Source connector and for managed Debezium CDC Source connectors, such as PostgreSQL, MySQL, SQL Server, and MariaDB.

For the Oracle XStream CDC Source connector, after.state.only doesn’t flatten records, and the unwrap alias isn’t reserved. To flatten records, add this SMT to the connector configuration.

Properties

The following table lists the configuration options for the ExtractNewRecordState SMT.

Name	Description	Type	Default	Importance
`delete.tombstone.handling.mode`	How to handle delete events and tombstones. Accepts `drop`, `tombstone`, `rewrite`, `rewrite-with-tombstone`, or `delete-to-tombstone`. Replaces `delete.handling.mode` and `drop.tombstones`. If you don’t set this option, the SMT uses `delete.handling.mode` and `drop.tombstones`.	string		medium
`delete.handling.mode`	Deprecated. Use `delete.tombstone.handling.mode` instead. How to handle delete events. Accepts `none`, `drop`, or `rewrite`.	string	`drop`	medium
`drop.tombstones`	Deprecated. Use `delete.tombstone.handling.mode` instead. Whether to drop tombstone records from the stream.	boolean	`true`	low
`route.by.field`	The name of a field in the row state whose value replaces the record’s topic name.	string		low
`add.fields`	A comma-separated list of change event metadata fields, such as `op`, `table`, or `source.ts_ms`, to add to the record value.	list		low
`add.fields.prefix`	The prefix added to the names of fields added by `add.fields`.	string	`__`	low
`add.headers`	A comma-separated list of change event metadata fields to add to the record headers.	list		low
`add.headers.prefix`	The prefix added to the names of headers added by `add.headers`.	string	`__`	low
`replace.null.with.default`	Whether to replace a null column value with the default value defined for the column in the source database.	boolean	`true`	low

Example

The following example flattens change events, rewrites delete events with a __deleted field, and adds the operation type to the record value. The example assumes that after.state.only is false, if the connector has that property.

"transforms": "flatten",
"transforms.flatten.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.flatten.delete.tombstone.handling.mode": "rewrite",
"transforms.flatten.add.fields": "op"

Predicates

Transformations can be configured with predicates so that the transformation is applied only to records which satisfy a condition. You can use predicates in a transformation chain and, when combined with the Kafka Connect Filter (Kafka) SMT Usage Reference for Confluent Cloud, predicates can conditionally filter out specific records. For details and examples, see Predicates.