Kafka Connect HeaderFrom SMT Usage Reference for Confluent Cloud

This topic describes the Apache Kafka® Single Message Transformation (SMT) org.apache.kafka.connect.transforms.HeaderFrom.

Description

The HeaderFrom SMT moves or copies fields in a record key or value into the record header. Use the concrete transformation type designed for the record key (org.apache.kafka.connect.transforms.HeaderFrom$Key) or value (org.apache.kafka.connect.transforms.HeaderFrom$Value).

Example

This configuration snippet shows how to use HeaderFrom to move the fields id and ts in the record value into the record header. The transform removes the fields from the record value. If you want to keep the fields in the record value, use the operation copy instead of move.

"transforms": "moveFieldsToHeader",
"transforms.moveFieldsToHeader.type": "org.apache.kafka.connect.transforms.HeaderFrom$Value",
"transforms.moveFieldsToHeader.fields": "id,ts",
"transforms.moveFieldsToHeader.headers": "record_id,event_timestamp",
"transforms.moveFieldsToHeader.operation": "move"

Before:

Record value

{
    "id": 24,
    "ts": 1626102708861,
    "name": "John Smith",
    "book": "Kafka: The Definitive Guide"
}

Record header
```
{
    "source": "RedditSource"
}
```

After:

Record value

{
    "name": "John Smith",
    "book": "Kafka: The Definitive Guide"
}

Record header

{
    "source": "RedditSource",
    "record_id": 24,
    "event_timestamp": 1626102708861
}

Tip

For additional examples, see HeaderFrom for managed connectors.

Properties

All properties are required.

Name	Description	Type	Default	Valid Values	Importance
`fields`	A comma-separated list of field names that the transform moves or copies to the record header.	list	None	non-empty list	high
`headers`	A comma-separated list of header names. The number of `headers` listed must be the same as the number of `fields` listed. The `headers` listed must be in the same order as the `fields` listed.	list	None	non-empty list	high
`operation`	`move`: Moves fields into the record header and removes these fields from the record value. `copy`: Adds the header entries but keeps the fields in the record value.	string	None	`move`, `copy`	high

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.