<commit_before>// Category: cdc-using
// Type: assembly

ifdef::community[]
[id="new-record-state-extraction"]
= New Record State Extraction
:toc:
:toc-placement: macro
:linkattrs:
:icons: font
:source-highlighter: highlight.js

toc::[]

[NOTE]
====
This single message transformation (SMT) is supported for only the SQL database connectors. For the MongoDB connector, see the {link-prefix}:{link-mongodb-event-flattening}[documentation for the MongoDB equivalent to this SMT].
====
endif::community[]

ifdef::product[]
[id="extracting-source-record-after-state-from-debezium-change-events"]
= Extracting source record `after` state from {prodname} change events
endif::product[]

A {prodname} data change event has a complex structure that provides a wealth of information. Kafka records that convey {prodname} change events contain all of this information. 
However, parts of a Kafka ecosystem expect Kafka records that provide a flat structure of field names and values. 
To provide this kind of record, {prodname} provides the `ExtractNewRecordState` single message transformation (SMT). Configure this transformation when consumers need Kafka records that have a format that is simpler than Kafka records that contain {prodname} change events. 

The `ExtractNewRecordState` transformation is a 
link:https://kafka.apache.org/documentation/#connect_transforms[Kafka Connect SMT].

ifdef::product[]
The transformation is available to only SQL database connectors. 

The following topics provide details: 

* xref:description-of-debezium-change-event-structure[]
* xref:behavior-of-debezium-extractnewrecordstate-transformation[]
* xref:configuration-of-extractnewrecordstate-transformation[]
* xref:example-of-adding-metadata-to-the-kafka-record[]
* xref:options-for-configuring-extractnewrecordstate-transformation[]

// Type: concept
[id="description-of-debezium-change-event-structure"]
== Description of {prodname} change event structure
endif::product[]

ifdef::community[]
== Change event structure
endif::community[]

{prodname} generates data change events that have a complex structure.
Each event consists of three parts:

* Metadata, which includes but is not limited to:

** The operation that made the change 
** Source information such as the names of the database and table where the change was made
** Time stamp for when the change was made
** Optional transaction information

* Row data before the change
* Row data after the change

For example, the structure of an `UPDATE` change event looks like this:

[source,json,indent=0]
----
{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}
----

ifdef::community[]
More details about change event structure are provided in 
xref:connectors/index.adoc[the documentation for each connector].
endif::community[]

This complex format provides the most information about changes happening in the system.
However, other connectors or other parts of the Kafka ecosystem usually expect the data in a simple format like this: 

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

To provide the needed Kafka record format for consumers, configure the `ExtractNewRecordState` SMT.

ifdef::community[]
== Behavior
endif::community[]

ifdef::product[]
// Type: concept
[id="behavior-of-debezium-extractnewrecordstate-transformation"]
== Behavior of {prodname} `ExtractNewRecordState` transformation
endif::product[]

link:https://github.com/debezium/debezium/blob/master/debezium-core/src/main/java/io/debezium/transforms/ExtractNewRecordState.java[The `ExtractNewRecordState` SMT] extracts the `after` field from a {prodname} change event in a Kafka record. The SMT replaces the original change event with only its `after` field to create a simple Kafka record. 

You can configure the `ExtractNewRecordState` SMT for a {prodname} connector or for a sink connector that consumes messages emitted by a {prodname} connector. The advantage of configuring `ExtractNewRecordState` for a sink connector is that records stored in Apache Kafka contain whole {prodname} change events. The decision to apply the SMT to a source or sink connector depends on your particular use case. 

You can configure the transformation to do any of the following: 

* Add metadata from the change event to the simplified Kafka record. The default behavior is that the SMT does not add metadata.

* Keep Kafka records that contain change events for `DELETE` operations in the stream. The default behavior is that the SMT drops Kafka records for `DELETE` operation change events because most consumers cannot yet handle them. 

A database `DELETE` operation causes {prodname} to generate two Kafka records: 

* A record that contains `"op": "d",` the `before` row data, and some other fields.
* A tombstone record that has the same key as the deleted row and a value of `null`. This record is a marker for Apache Kafka. It indicates that 
link:https://kafka.apache.org/documentation/#compaction[log compaction] can remove all records that have this key. 

Instead of dropping the record that contains the `before` row data, you can configure the `ExtractNewRecordData` SMT to do one of the following: 

* Keep the record in the stream and edit it to have only the `"value": "null"` field.
 
* Keep the record in the stream and edit it to have a `value` field that contains the key/value pairs that were in the `before` field with an added `"__deleted": "true"` entry.

Similary, instead of dropping the tombstone record, you can configure the `ExtractNewRecordData` SMT to keep the tombstone record in the stream. 

ifdef::community[]
== Configuration
endif::community[]

ifdef::product[]
// Type: concept
[id="configuration-of-extractnewrecordstate-transformation"]
== Configuration of `ExtractNewRecordState` transformation
endif::product[]

Configure the {prodname} `ExtractNewRecordState` SMT in a Kafka Connect source or sink connector. Add the SMT configuration details to your connector's configuration. To obtain the default behavior, in a `.properties` file, you would specify something like the following: 

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
----

As for any Kafka Connect connector configuration, you can set `transforms=` to multiple, comma-separated, SMT aliases in the order in which you want Kafka Connect to apply the SMTs. 

The following `.properties` example sets several `ExtractNewRecordState` options: 

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=rewrite
transforms.unwrap.add.fields=table,lsn
----

`drop-tombstones=false`:: Keeps tombstone records for `DELETE` operations in the event stream. 

`delete-handling-mode=rewrite`:: For `DELETE` operations, edits the Kafka record by flattening the `value` field that was in the change event. The `value` field directly contains the key/value pairs that were in the `before` field. The SMT adds `__deleted` and sets it to `true`, for example:   
+<commit_msg>DBZ-1956 Corrected property names by replaceing hyphens with dots in two instances<commit_after>// Category: cdc-using
// Type: assembly

ifdef::community[]
[id="new-record-state-extraction"]
= New Record State Extraction
:toc:
:toc-placement: macro
:linkattrs:
:icons: font
:source-highlighter: highlight.js

toc::[]

[NOTE]
====
This single message transformation (SMT) is supported for only the SQL database connectors. For the MongoDB connector, see the {link-prefix}:{link-mongodb-event-flattening}[documentation for the MongoDB equivalent to this SMT].
====
endif::community[]

ifdef::product[]
[id="extracting-source-record-after-state-from-debezium-change-events"]
= Extracting source record `after` state from {prodname} change events
endif::product[]

A {prodname} data change event has a complex structure that provides a wealth of information. Kafka records that convey {prodname} change events contain all of this information. 
However, parts of a Kafka ecosystem expect Kafka records that provide a flat structure of field names and values. 
To provide this kind of record, {prodname} provides the `ExtractNewRecordState` single message transformation (SMT). Configure this transformation when consumers need Kafka records that have a format that is simpler than Kafka records that contain {prodname} change events. 

The `ExtractNewRecordState` transformation is a 
link:https://kafka.apache.org/documentation/#connect_transforms[Kafka Connect SMT].

ifdef::product[]
The transformation is available to only SQL database connectors. 

The following topics provide details: 

* xref:description-of-debezium-change-event-structure[]
* xref:behavior-of-debezium-extractnewrecordstate-transformation[]
* xref:configuration-of-extractnewrecordstate-transformation[]
* xref:example-of-adding-metadata-to-the-kafka-record[]
* xref:options-for-configuring-extractnewrecordstate-transformation[]

// Type: concept
[id="description-of-debezium-change-event-structure"]
== Description of {prodname} change event structure
endif::product[]

ifdef::community[]
== Change event structure
endif::community[]

{prodname} generates data change events that have a complex structure.
Each event consists of three parts:

* Metadata, which includes but is not limited to:

** The operation that made the change 
** Source information such as the names of the database and table where the change was made
** Time stamp for when the change was made
** Optional transaction information

* Row data before the change
* Row data after the change

For example, the structure of an `UPDATE` change event looks like this:

[source,json,indent=0]
----
{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}
----

ifdef::community[]
More details about change event structure are provided in 
xref:connectors/index.adoc[the documentation for each connector].
endif::community[]

This complex format provides the most information about changes happening in the system.
However, other connectors or other parts of the Kafka ecosystem usually expect the data in a simple format like this: 

[source,json,indent=0]
----
{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}
----

To provide the needed Kafka record format for consumers, configure the `ExtractNewRecordState` SMT.

ifdef::community[]
== Behavior
endif::community[]

ifdef::product[]
// Type: concept
[id="behavior-of-debezium-extractnewrecordstate-transformation"]
== Behavior of {prodname} `ExtractNewRecordState` transformation
endif::product[]

link:https://github.com/debezium/debezium/blob/master/debezium-core/src/main/java/io/debezium/transforms/ExtractNewRecordState.java[The `ExtractNewRecordState` SMT] extracts the `after` field from a {prodname} change event in a Kafka record. The SMT replaces the original change event with only its `after` field to create a simple Kafka record. 

You can configure the `ExtractNewRecordState` SMT for a {prodname} connector or for a sink connector that consumes messages emitted by a {prodname} connector. The advantage of configuring `ExtractNewRecordState` for a sink connector is that records stored in Apache Kafka contain whole {prodname} change events. The decision to apply the SMT to a source or sink connector depends on your particular use case. 

You can configure the transformation to do any of the following: 

* Add metadata from the change event to the simplified Kafka record. The default behavior is that the SMT does not add metadata.

* Keep Kafka records that contain change events for `DELETE` operations in the stream. The default behavior is that the SMT drops Kafka records for `DELETE` operation change events because most consumers cannot yet handle them. 

A database `DELETE` operation causes {prodname} to generate two Kafka records: 

* A record that contains `"op": "d",` the `before` row data, and some other fields.
* A tombstone record that has the same key as the deleted row and a value of `null`. This record is a marker for Apache Kafka. It indicates that 
link:https://kafka.apache.org/documentation/#compaction[log compaction] can remove all records that have this key. 

Instead of dropping the record that contains the `before` row data, you can configure the `ExtractNewRecordData` SMT to do one of the following: 

* Keep the record in the stream and edit it to have only the `"value": "null"` field.
 
* Keep the record in the stream and edit it to have a `value` field that contains the key/value pairs that were in the `before` field with an added `"__deleted": "true"` entry.

Similary, instead of dropping the tombstone record, you can configure the `ExtractNewRecordData` SMT to keep the tombstone record in the stream. 

ifdef::community[]
== Configuration
endif::community[]

ifdef::product[]
// Type: concept
[id="configuration-of-extractnewrecordstate-transformation"]
== Configuration of `ExtractNewRecordState` transformation
endif::product[]

Configure the {prodname} `ExtractNewRecordState` SMT in a Kafka Connect source or sink connector. Add the SMT configuration details to your connector's configuration. To obtain the default behavior, in a `.properties` file, you would specify something like the following: 

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
----

As for any Kafka Connect connector configuration, you can set `transforms=` to multiple, comma-separated, SMT aliases in the order in which you want Kafka Connect to apply the SMTs. 

The following `.properties` example sets several `ExtractNewRecordState` options: 

[source]
----
transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=rewrite
transforms.unwrap.add.fields=table,lsn
----

`drop.tombstones=false`:: Keeps tombstone records for `DELETE` operations in the event stream. 

`delete.handling.mode=rewrite`:: For `DELETE` operations, edits the Kafka record by flattening the `value` field that was in the change event. The `value` field directly contains the key/value pairs that were in the `before` field. The SMT adds `__deleted` and sets it to `true`, for example:   
+
{'fraction_non_alphanumeric': 0.09072179108577708, 'fraction_numerical': 0.0021976512602156447, 'mean_word_length': 5.12620950778292, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 6, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3403392', 'n_tokens_mistral': 4043, 'n_tokens_neox': 3951, 'n_words': 1673}