Kafka Connect ExtractTopic SMT Usage Reference for Confluent Cloud๏ƒ

The io.confluent.connect.transforms.ExtractTopic Confluent Single Message Transformation (SMT) extracts data from a message and uses it as the topic name. Supported extraction sources include:

  • Full payload: Uses the entire key or value, which must be a string.

  • Nested field: Extracts a specific field from a complex data structure such as Map or Struct.

  • Message header: Uses the entire string value from a header.

Use the concrete transformation type required for your target location:

  • io.confluent.connect.transforms.ExtractTopic$Key: Extracts from the record key.

  • io.confluent.connect.transforms.ExtractTopic$Value: Extracts from the record value.

  • io.confluent.connect.transforms.ExtractTopic$Header: Extracts from a record header.

Installation๏ƒ

This transformation is developed by Confluent and does not ship by default with Apache Kafkaยฎ or Confluent Cloud. You can install this transformation using the confluent connect plugin install command:

confluent connect plugin install confluentinc/connect-transforms:latest

Examples๏ƒ

The configuration snippet below shows how to use and configure the ExtractTopic SMT.

"transforms": "KeyExample", "ValueFieldExample", "KeyFieldExample", "FieldJsonPathExample", "HeaderExample",

Use the key of the message as the topic name.

"transforms.KeyExample.type": "io.confluent.connect.transforms.ExtractTopic$Key"

Extract a required field named f2 from the value, and use it as the topic name.

"transforms.ValueFieldExample.type": "io.confluent.connect.transforms.ExtractTopic$Value",
"transforms.ValueFieldExample.field": "f2"

Extract a field named f3 from the key, and use it as the topic name. If the field is null or missing, leave the topic name as-is.

"transforms.KeyFieldExample.type": "io.confluent.connect.transforms.ExtractTopic$Value",
"transforms.KeyFieldExample.field": "f3",
"transforms.KeyFieldExample.skip.missing.or.null": "true"

Extract the value of a field named f3 in the f1 field in the key, and use it as the topic name. Here the format of the field is defined with JSON Path (e.g., ["f1"]["f3"]). If the field is null or missing, leave the topic name as-is.

"transforms.FieldJsonPathExample.type": "io.confluent.connect.transforms.ExtractTopic$Value",
"transforms.FieldJsonPathExample.field": "$["f1"]["f3"]",
"transforms.FieldJsonPathExample.field.format": "JSON_PATH",
"transforms.FieldJsonPathExample.skip.missing.or.null": "true"

Extract the value of a message header (as a string) with key h1 (required) and use it as the topic name.

"transforms.HeaderExample.type": "io.confluent.connect.transforms.ExtractTopic$Header",
"transforms.HeaderExample.field": "h1",
"transforms.FieldJsonPathExample.skip.missing.or.null=true"

Tip

For additional examples, see ExtractTopic for managed connectors.

Properties๏ƒ

Name

Description

Type

Default

Valid Values

Importance

field

Field name to use as the topic name. If left blank, the entire key or value is used (and assumed to be a string).

string

โ€œโ€

medium

field.format

Specify field path format. Currently two formats are supported: JSON_PATH and PLAIN. If set to JSON_PATH, the transformer interprets the field with a JSON path interpreter, which supports nested field extraction. If left blank or set to PLAIN, the transformer evaluates the field configuration as a non-nested field name. When using ExtractTopic$Header, only the default PLAIN format can be used, which extracts the header value as a string.

string

โ€œPLAINโ€

โ€œJSON_PATHโ€, โ€œPLAINโ€

medium

skip.missing.or.null

How to handle missing fields and null fields, keys, and values. By default, this transformation throws an exception if a field defined in the field configuration is missing or null, or if no field is specified but the messageโ€™s key or value is null. If this configuration is set to true, the transformation instead silently ignores these conditions and allows the record to pass through unaltered.

boolean

false

low

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.