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

The ExtractChangedRecordState SMT (io.debezium.transforms.ExtractChangedRecordState) inspects Debezium change events and attaches Apache Kafka® message headers identifying modified and unmodified fields during database UPDATE operations.

Description

Use this SMT for downstream consumers to detect field-level changes without comparing the full before and after structures.

For an UPDATE event, the SMT compares the before and after field values to populate the headers.
For INSERT, DELETE, and read or snapshot events generated when Debezium performs its initial data snapshot, the SMT adds the headers configured through header.changed.name and header.unchanged.name as empty lists.

For complete details, see the official Debezium ExtractChangedRecordState SMT documentation.

For the full list of available SMTs, see Kafka Connect Single Message Transformation Reference for Confluent Cloud.

Limitations

The ExtractChangedRecordState SMT is available only for managed Debezium change data capture (CDC) Source connectors, such as PostgreSQL, MySQL, SQL Server, and MariaDB, and the Oracle XStream CDC Source connector. Field-level change detection applies to UPDATE events only.

Properties

To add headers, set header.changed.name, header.unchanged.name, or both. The following table lists the configuration options for the ExtractChangedRecordState SMT.

Name	Description	Type	Importance
`header.changed.name`	The name of the Kafka message header that stores a comma-separated list of the fields the database operation changed.	string	low
`header.unchanged.name`	The name of the Kafka message header that stores a comma-separated list of the fields that remained unchanged after the database operation.	string	low

Example

The following example adds a Changed header and an Unchanged header to each record.

"transforms": "changes",
"transforms.changes.type": "io.debezium.transforms.ExtractChangedRecordState",
"transforms.changes.header.changed.name": "Changed",
"transforms.changes.header.unchanged.name": "Unchanged"

Note

This SMT adds its output to the record headers, not to the record value.

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.