<a id="insertfield"></a>

# 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`.

```json
"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.

```json
"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"}`

## Properties

| Name                        | Description                                                                                                                                                                        | Type    | Default   | Valid values   | Importance   |
|-----------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------|-----------|----------------|--------------|
| `offset.field`              | Field name for Kafka offset. This is only applicable to sink<br/>connectors. Suffix with `!` to make this a required field, or<br/>`?` to keep it optional (the default).          | String  | Null      |                | Medium       |
| `partition.field`           | Field name for Kafka partition. Suffix with `!` to make this a<br/>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<br/>are null with the default value. When set to `true`, the SMT<br/>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`.<br/>Suffix with `!` to make this a required field, or `?` to<br/>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<br/>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<br/>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](filter-ak.md#ak-filter), predicates
can conditionally filter out specific records. For details and examples, see [Predicates](filter-ak.md#predicates).