Kafka Connect HeaderFrom SMT Usage Reference for Confluent Cloud or Confluent Platform

The following provides usage information for the Apache Kafka® SMT org.apache.kafka.connect.transforms.HeaderFrom.

Caution

HeaderFrom is not currently available for managed connectors.

Description

Moves or copies fields in a record key or value into the record’s 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’s 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,
}

Properties

All properties are required.

Name	Description	Type	Valid Values	Importance
`fields`	A comma-separated list of field names that the transform moves or copies to the record header.	list	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	non-empty list	high
`operation`	The `move` operation moves fields into the record header and removes these fields from the record value. The `copy` operation adds the header entries but keeps the fields in the record value.	string	`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 or Confluent Platform, predicates can conditionally filter out specific records. For details and examples, see Predicates.