Kafka Connect Filter (Kafka) SMT Usage Reference for Confluent Cloud

The org.apache.kafka.connect.transforms.Filter Apache Kafka® Single Message Transformation (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: Conditional routing and transformation using predicates

A source connector produces records to many different topics. This example demonstrates how to:

Drop all records in the foo topic.
Apply the ExtractField transformation with the field name other_field to records in every topic except bar.

Step 1:

Use the Filter transformation to drop records in the foo topic. Configure it with the TopicNameMatches predicate, which matches topic names against a Java regular expression.

"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"

Step 2:

Because the TopicNameMatches predicate matches by inclusion rather than exclusion, you cannot explicitly target every topic except bar using only the pattern.

Instead, use the transformation’s implicit negate property. Setting negate to true inverts the predicate’s behavior, ensuring the transformation runs only when the predicate evaluates to false.

"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"

With negate=true, ExtractField runs on every record whose topic name doesn’t match bar.

Example 2: Excluding topics through regex prefix matching

This example demonstrates how to use the TopicNameMatches predicate with negate=true to apply ExtractField only to topics that don’t match a given prefix.

"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 predicates.has-my-prefix keys configure a TopicNameMatches predicate with the pattern my-prefix-.*. Because negate=true inverts the predicate, t2 runs only on records whose topic name doesn’t start with my-prefix-.

Example 3: Filtering records based on header presence

This example demonstrates how to use the HasHeaderKey predicate to drop records that carry a header named dropped. The predicate’s name property specifies the header key to match:

"transforms": "Filter",
"transforms.Filter.type": "org.apache.kafka.connect.transforms.Filter",
"transforms.Filter.predicate": "hasDroppedHeader",

"predicates": "hasDroppedHeader",
"predicates.hasDroppedHeader.type": "org.apache.kafka.connect.transforms.predicates.HasHeaderKey",
"predicates.hasDroppedHeader.name": "dropped"

With this configuration, the Filter transformation drops any record that carries a header named dropped, regardless of the header’s value.

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	Configuration property	Description	Type	Valid values	Importance
`TopicNameMatches`	`pattern`	A predicate which is true for records with a topic name that matches the configured regular expression.	String	Non-empty string, valid regex	Medium
`HasHeaderKey`	`name`	A predicate which is true for records with at least one header with the configured name.	String	Non-empty string	Medium
`RecordIsTombstone`		A predicate which is true for records that are tombstones (that is, records with a null value).			Medium