Kafka Connect InsertField SMT Usage Reference for Confluent Cloud

The org.apache.kafka.connect.transforms.InsertField Apache Kafka® Single Message Transformation (SMT) inserts fields using attributes from the record metadata or a configured static value.

Use the concrete transformation type designed for the record key (org.apache.kafka.connect.transforms.InsertField$Key) or value (org.apache.kafka.connect.transforms.InsertField$Value).

If the record’s key or value is null, the SMT skips the transformation and passes the record through unchanged.

The partition.field and timestamp.field properties depend on metadata that a record might not carry. The SMT inserts a partition field only when the record has a partition, and a timestamp field only when the record has a timestamp. If that metadata is missing, the SMT skips the corresponding field.

Examples

The following two configuration snippet examples show how to use the InsertField SMT.

Insert a static field

To insert a static field, use transforms.InsertField.static.field and transforms.InsertField.static.value to include the defined static field and value. In this example, the field is MessageSource with the value Kafka Connect framework.

"transforms": "InsertField",
"transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.static.field": "MessageSource",
"transforms.InsertField.static.value": "Kafka Connect framework"

Before: {"author": "Philip K. Dick", "character": "Palmer Eldritch"}

After: {"author": "Philip K. Dick", "character": "Palmer Eldritch", "MessageSource": "Kafka Connect framework"}

Insert a field into the record key

To insert a field into the record key instead of the value, use the InsertField$Key transformation type with any of the field-name properties listed in the Properties table, such as topic.field. InsertField has no separate key.field property.

"transforms": "InsertField",
"transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Key",
"transforms.InsertField.topic.field": "SourceTopic"

Before: key {"id": 42}

After: key {"id": 42, "SourceTopic": "orders"}

Tip

For additional examples, see Insert Field for managed connectors.

Properties

Name	Description	Type	Default	Valid values	Importance
`offset.field`	Field name for Kafka offset. This is only applicable to sink connectors. Suffix with `!` to make this a required field, or `?` to keep it optional (the default).	String	Null		Medium
`partition.field`	Field name for Kafka partition. Suffix with `!` to make this a required field, or `?` to keep it optional (the default).	String	Null		Medium
`replace.null.with.default`	Whether to replace fields that have a default value and that are null with the default value. When set to `true`, the SMT uses the default value. Otherwise, it uses null.	Boolean	true	true or false	Medium
`static.field`	Field name for the static value defined in `static.value`. Suffix with `!` to make this a required field, or `?` to keep it optional (the default).	String	Null		Medium
`static.value`	The static value for the field specified in `static.field`.	String	Null		Medium
`timestamp.field`	Field name for record timestamp. Suffix with `!` to make this a required field, or `?` to keep it optional (the default).	String	Null		Medium
`topic.field`	Field name for Kafka topic. Suffix with `!` to make this a required field, or `?` to keep it optional (the default).	String	Null		Medium

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.