# 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](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command: ```properties confluent connect plugin install confluentinc/connect-transforms:latest ``` ## Examples The configuration snippet below shows how to use and configure the `ExtractTopic` SMT. ```json "transforms": "KeyExample", "ValueFieldExample", "KeyFieldExample", "FieldJsonPathExample", "HeaderExample", ``` Use the key of the message as the topic name. ```json "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. ```json "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. ```json "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. ```json "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. ```json "transforms.HeaderExample.type": "io.confluent.connect.transforms.ExtractTopic$Header", "transforms.HeaderExample.field": "h1", "transforms.FieldJsonPathExample.skip.missing.or.null=true" ``` ## 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](filter-ak.md#ak-filter), predicates can conditionally filter out specific records. For details and examples, see [Predicates](filter-ak.md#predicates).