# 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: 1. Drop all records in the `foo` topic. 2. 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. ```json "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`. ```json "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. ```json "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: ```json "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. ## Predicate properties | Name | Configuration property | Description | Type | Default | 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 |