Description
Set the schema name, version or both on the record’s key (org.apache.kafka.connect.transforms.SetSchemaMetadata$Key) or value (org.apache.kafka.connect.transforms.SetSchemaMetadata$Value) schema.
This SMT can be used to set the schema name, version, or both on Connect records. Since the schema name includes the namespace, a common SetSchemaMetadata SMT use case is to change the name and namespace of the schemas used in Apache Kafka® record keys and values.
Use cases:
Source connectors generate records with schemas defined by the connector, and often generate the names and versions of key and value schemas based upon source-specific information. For example, a database source connector might use the name of the table from which rows are read in the schema names. If these generated schema names do not adhere to the naming convention you need, you can use this SMT to override the generated names of the key and/or value schemas in the source records produced by the source connector.
Sink connectors that consume records from Kafka topics may the names of the schemas to indicate how the connector maps the Kafka records into the external system. If the names of the schemas used in the record keys and values don’t result in the desired mapping, you can use this SMT to change the name of the key and/or value schemas in the consumed source records, before those records are processed by the sink connector.
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 Apache Kafka® Filter, predicates can conditionally filter out specific records.
Predicates are specified in the connector configuration. The following properties are used:
predicates: A set of aliases for predicates applied to one or more transformations.predicates.$alias.type: Fully qualified class name for the predicate.predicates.$alias.$predicateSpecificConfig: Configuration properties for the predicate.
All transformations have the implicit config properties predicate and
negate. A predicular predicate is associated with a transformation by
setting the transformation’s predicate configuration to the predicate’s alias.
The predicate’s value can be reversed using the negate configuration
property.
Kafka Connect includes the following predicates:
org.apache.kafka.connect.predicates.TopicNameMatches: Matches records in a topic with a name matching a particular Java regular expression.org.apache.kafka.connect.predicates.HasHeaderKey: Matches records which have a header with the given key.org.apache.kafka.connect.predicates.RecordIsTombstone: Matches tombstone records (that is, records with a null value).
Predicate Examples
Example 1:
You have a source connector that produces records to many different topics and
you want to do the following:
- Filter out the records in the
foo topic entirely.
- Apply the
ExtractField transformation with the field name other_field to records in all topics, except the topic bar.
To do this, you need to first filter out the records destined for the topic
foo. The Filter transformation removes records from further processing.
Next, you use the TopicNameMatches predicate to apply the transformation
only to records in topics which match a certain regular expression. The only
configuration property for TopicNameMatches is a Java regular expression
used as a pattern for matching against the topic name. The following example
shows this configuration:
transforms=Filter
transforms.Filter.type=org.apache.kafka.connect.transforms.Filter
transforms.Filter.predicate=IsFoo
predicates=IsFoo
predicates.IsFoo.type=org.apache.kafka.connect.predicates.TopicNameMatches
predicates.IsFoo.pattern=foo
Using this configuration, ExtractField is then applied only when the topic
name of the record is not bar. The reason you can’t use TopicNameMatches
directly is because it would apply the transformation to matching topic names,
not topic names which do not match. The transformation’s implicit negate
configuration properties inverts the set of records which a predicate matches.
This configuration addition is shown below:
transforms=Filter,Extract
transforms.Filter.type=org.apache.kafka.connect.transforms.Filter
transforms.Filter.predicate=IsFoo
transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=other_field
transforms.Extract.predicate=IsBar
transforms.Extract.negate=true
predicates=IsFoo,IsBar
predicates.IsFoo.type=org.apache.kafka.connect.predicates.TopicNameMatches
predicates.IsFoo.pattern=foo
predicates.IsBar.type=org.apache.kafka.connect.predicates.TopicNameMatches
predicates.IsBar.pattern=bar
Example 2:
The following configuration shows how to use a predicate in a transformation
chain with the ExtractField transformation and the negate=true
configuration property:
transforms=t2
transforms.t2.predicate=has-my-prefix
transforms.t2.negate=true
transforms.t2.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.t2.field=c1
predicates=has-my-prefix
predicates.has-my-prefix.type=org.apache.kafka.connect.predicates.TopicNameMatch
predicates.has-my-prefix.pattern=my-prefix-.*
The transform t2 is only applied when the predicate has-my-prefix is
false (using the negate=true parameter). The predicate is configured by the
keys with prefix predicates.has-my-prefix. The predicate class is
org.apache.kafka.connect.predicates.TopicNameMatch and it’s pattern
parameter has the value my-prefix-.* . With this configuration, the
transformation is applied only to records where the topic name does not
start with my-prefix-.
Tip
The benefit of defining the predicate separately from the transform is it
makes it easier to apply the same predicate to multiple transforms. For
example, you can have one set of transforms use one predicate and another set
of transforms use the same predicate for negation.