Kafka Connect Filter (Kafka) SMT Usage Reference for Confluent Cloud
The org.apache.kafka.connect.transforms.Filter Apache Kafka® Single Message Transform (SMT) drops records based on a predicate condition.
Description
The Filter SMT drops all records, filtering them from subsequent transformations in the chain. It is used conditionally to filter out records matching (or not matching) a predicate.
Predicates
A predicate is a condition evaluated against each record. A transformation configured with a predicate is applied only to records that satisfy it. Transformations can be combined with predicates in a chain and, when used with the Apache Kafka® Filter, predicates can conditionally filter out 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 configuration properties predicate and negate. A 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.
The following predicates are available:
org.apache.kafka.connect.transforms.predicates.TopicNameMatches: Matches records in a topic with a name matching a particular Java regular expression.org.apache.kafka.connect.transforms.predicates.HasHeaderKey: Matches records that have a header with the given key.org.apache.kafka.connect.transforms.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
footopic entirely.Apply the
ExtractFieldtransformation with the field nameother_fieldto records in all topics, except the topicbar.
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 that 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.transforms.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 that do not match. The transformation’s implicit negate configuration properties invert the set of records that a predicate matches. This configuration addition is shown in the following example:
"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.transforms.predicates.TopicNameMatches",
"predicates.IsFoo.pattern": "foo",
"predicates.IsBar.type": "org.apache.kafka.connect.transforms.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.transforms.predicates.TopicNameMatches",
"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.transforms.predicates.TopicNameMatches and its 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 that 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.
For additional examples, see Filter (Apache Kafka) for managed connectors.
Predicate properties
Name | Description | Type | Default | Valid Values | Importance |
|---|---|---|---|---|---|
| A predicate which is true for records with a topic name that matches the configured regular expression. | String | Non-empty string, valid regex | Medium | |
| A predicate which is true for records with at least one header with the configured name. | String | Non-empty string | Medium | |
| A predicate which is true for records that are tombstones (that is, records with a null value). | Medium |