Home

Awesome

Aiven Transformations for Apache Kafka® Connect

A collection of Single Message Transformations (SMTs) for Apache Kafka Connect.

Transformations

See the Kafka documentation for more details about configuring transformations or demo on how to install transforms.

ExtractTimestamp

This transformation replaces the original record's timestamp with a value taken from the the record.

The transformation:

Exists in two variants:

The transformation defines the following configurations:

Here's an example of this transformation configuration:

transforms=ExtractTimestampFromValueField
transforms.ExtractTimestampFromValueField.type=io.aiven.kafka.connect.transforms.ExtractTimestamp$Value
transforms.ExtractTimestampFromValueField.field.name=inner_field_name

ExtractTopic

This transformation extracts a string value from the record and use it as the topic name.

The transformation can use either the whole key or value (in this case, it must have INT8, INT16, INT32, INT64, FLOAT32, FLOAT32, BOOLEAN, or STRING type; or related classes) or a field in them (in this case, it must have STRUCT type and the field's value must be INT8, INT16, INT32, INT64, FLOAT32, FLOAT32, BOOLEAN, or STRING; or related).

It supports fields with (e.g. Avro) or without schema (e.g. JSON).

Exists in two variants:

The transformation defines the following configurations:

Here is an example of this transformation configuration:

transforms=ExtractTopicFromValueField
transforms.ExtractTopicFromValueField.type=io.aiven.kafka.connect.transforms.ExtractTopic$Value
transforms.ExtractTopicFromValueField.field.name=inner_field_name

Hash

This transformation replaces a string value with its hash.

The transformation can hash either the whole key or value (in this case, it must have STRING type) or a field in them (in this case, it must have STRUCT type and the field's value must be STRING).

Exists in two variants:

The transformation defines the following configurations:

Here is an example of this transformation configuration:

transforms=HashEmail
transforms.HashEmail.type=io.aiven.kafka.connect.transforms.Hash$Value
transforms.HashEmail.field.name=email
transforms.HashEmail.function=sha1

TombstoneHandler

This transformation manages tombstone records, i.e. records with the entire value field being null.

The transformation defines the following configurations:

Here is an example of this transformation configuration:

transforms=TombstoneHandler
transforms.TombstoneHandler.type=io.aiven.kafka.connect.transforms.TombstoneHandler
transforms.TombstoneHandler.behavior=drop_silent

ConcatFields

This transformation adds a new field to the message with a key of type string and a value of string which is the concatenation of the requested fields.

Exists in two variants:

The transformation defines the following configurations:

Here is an example of this transformation configuration:

transforms=ConcatFields
transforms.ConcatFields.type=io.aiven.kafka.connect.transforms.ConcatFields$Value
transforms.ConcatFields.field.names=test,foo,bar,age
transforms.ConcatFields.output.field.name="combined"
transforms.ConcatFields.delimiter="-"
transforms.ConcatFields.field.replace.missing="*"

MakeTombstone

This transformation converts a record into a tombstone by setting its value and value schema to null.

It can be used together with predicates, for example, to create a tombstone event from a delete event produced by a source connector.

Here is an example of this transformation configuration:

transforms=MakeTombstone
transforms.MakeTombstone.type=io.aiven.kafka.connect.transforms.MakeTombstone

FilterByFieldValue

This transformation allows filtering records based either on a specific field or whole value and a matching expected value or regex pattern.

Here is an example of this transformation configuration:

transforms=Filter
transforms.Filter.type=io.aiven.kafka.connect.transforms.FilterByFieldValue
transforms.Filter.field.name=<field_name>
transforms.Filter.field.value=<field_value>
transforms.Filter.field.value.pattern=<regex_pattern>
transforms.Filter.field.value.matches=<true|false>

If field.name is empty, the whole value is considered for filtering.

Either field.value or field.value.pattern must be defined to apply filter.

Only, string, numeric and boolean types are considered for matching purposes, other types are ignored.

ExtractTopicFromSchemaName

This transformation checks the schema name and if it exists uses it as the topic name.

Currently this transformation only has implementation for record value schema name. Key schema name is not implemented.

By default (if schema.name.topic-map or the chema.name.regex is not set) transformation uses the content of the schema.name field.

The transformation defines the following optional configurations which can be used to tamper the schema.name:

Here is an example of this transformation configuration (using :schema.name.topic-map)

transforms=ExtractTopicFromSchemaName
transforms.ExtractTopicFromSchemaName.type=io.aiven.kafka.connect.transforms.ExtractTopicFromSchemaName$Value
transforms.ExtractTopicFromSchemaName.schema.name.topic-map=com.acme.schema.SchemaNameToTopic1:TheNameToReplace1,com.acme.schema.SchemaNameToTopic2:TheNameToReplace2

And here is an example of this transformation configuration (using :schema.name.regex)

transforms=ExtractTopicFromValueSchema
transforms.ExtractTopicFromValueSchema.type=io.aiven.kafka.connect.transforms.ExtractTopicFromSchemaName$Value
transforms.ExtractTopicFromValueSchema.schema.name.regex=(?:[.]|^)([^.]*)$

License

This project is licensed under the Apache License, Version 2.0.

Trademarks

Apache Kafka and Apache Kafka Connect are either registered trademarks or trademarks of the Apache Software Foundation in the United States and/or other countries.