The following provides usage information for the Confluent SMT
io.confluent.connect.transforms.Filter. Include or drop records that match the
Include or drop records that match the
filter.condition is a predicate specifying JSON Path that is applied to each record processed, and when this predicate successfully matches the record is either included (when
filter.type=include) or excluded (when
missing.or.null.behavior property defines how the transform behaves when a record does not have the field(s) used in the filter condition predicate. By default the behavior is to fail. This property can also be set to include or exclude the record that is missing the predicate’s field(s).
Use the transformation type designed for the record key
io.confluent.connect.transforms.Filter$Key) or value
This transformation is developed by Confluent and does not ship by default with Kafka or Confluent Platform. You can install this transformation using the Confluent Hub Client:
confluent-hub install confluentinc/connect-transforms:latest
The configuration snippet below shows how to use and configure the Confluent Filter SMT.
filter.condition is a predicate specifying JSON path. For more
information on JSON path format, see https://github.com/json-path/JsonPath.
filter.condition specified in the following examples is satisfied when the
'nestedKey' in field
'key' is equal to either
Defines the configuration for
satisfying the filter condition. The filter condition is applied on record
fail behavior in the configuration specifies to throw an
exception and fail the connector task when the record does not have the
field(s) used in the filter condition.
"transforms": "filterExample1", "transforms.filterExample1.type": "io.confluent.connect.transforms.Filter$Value", "transforms.filterExample1.filter.condition": "$.key[?(@.nestedKey == "value1" || @.nestedKey == "value2")]", "transforms.filterExample1.filter.type": "include", "transforms.filterExample1.missing.or.null.behavior": "fail"
Defines the configuration for
excluding records that satisfy the filter condition. The filter condition is applied on record
include behavior in the configuration specifies to pass the
record through when the record does not have the field(s) used in the filter
"transforms": "filterExample2", "transforms.filterExample2.type": "io.confluent.connect.transforms.Filter$Key", "transforms.filterExample2.filter.condition": "$.key[?(@.nestedKey == "value1" || @.nestedKey == "value2")]", "transforms.filterExample2.filter.type": "exclude", "transforms.filterExample2.missing.or.null.behavior": "include"
This example includes syntax to filter on a simple schema:
"transforms":"filterExample3", "transforms.filterExample3.type":"io.confluent.connect.transforms.Filter$Value", "transforms.filterExample3.filter.condition.": "$[?(@.fieldToFilter == 'valueToMatch')]", "transforms.filterExample3.filter.type":"exclude", "transforms.filterExample3.missing.or.null.behavior":"include",
For additional examples, see Filter (Confluent) for fully-managed connectors.
||Specifies the criteria used to match records to be included or excluded by this transformation. Use JSON Path predicate notation defined in: https://github.com/json-path/JsonPath.||string||high|
||Specifies the action to perform with records that match the
||Specifies the behavior when the record does not have the field(s) used
||[fail, include, exclude]||medium|