KafkaIQ

4.1 – Single Message Transforms (SMTs)

Overview

This module is the authoritative explanation of Single Message Transforms (SMTs) in Kafka Connect. By the end, you will understand:

  • What SMTs are and when to use them.
  • Built-in SMTs provided by Kafka Connect.
  • How to chain multiple SMTs.
  • Key vs. value transformations.
  • Common real-world SMT patterns.

All SMT concepts live here. Other modules will reference this when discussing connector configuration or custom connector development, but will not re-explain SMT mechanics.

What Are Single Message Transforms (SMTs)?

Definition

Single Message Transforms (SMTs) are lightweight, record-level transformations applied to data as it flows through Kafka Connect—before source records are written to Kafka or after sink records are read from Kafka.

Purpose

SMTs enable simple data transformations without requiring:

  • Custom connector code.
  • External stream processing frameworks (e.g., Kafka Streams, Flink).

Characteristics

  • Stateless: Each record is transformed independently (no aggregation, windowing, or joins).
  • Synchronous: Transformations happen inline during data ingestion/egress.
  • Chainable: Multiple SMTs can be applied in sequence.

Analogy: SMTs are like filters in a data pipeline—they modify records as they pass through, without altering the source or sink systems.

When to Use SMTs

✅ Use SMTs For

  1. Field Renaming: Change field names to match schema conventions.
  2. Field Filtering: Remove sensitive or unnecessary fields.
  3. Data Type Conversion: Convert strings to integers, timestamps to epoch millis, etc.
  4. Routing: Send records to different topics based on field values.
  5. Timestamp Injection: Add Kafka record timestamps to message payloads.
  6. Key Extraction: Extract a field from the value to use as the record key.

❌ Avoid SMTs For

  1. Stateful Transformations: Aggregations, joins, windowing (use Kafka Streams instead).
  2. Complex Business Logic: Multi-step transformations, enrichment from external systems (use custom code).
  3. High-Latency Operations: External API calls, database lookups (degrades throughput).

Rule of Thumb: If the transformation is simple, stateless, and record-scoped, use an SMT. Otherwise, use a stream processor or custom code.

How SMTs Work

Data Flow

For Source Connectors

External System → Source Task → [SMTs] → Kafka Topic

SMTs are applied after the source task reads data but before it writes to Kafka.

For Sink Connectors

Kafka Topic → [SMTs] → Sink Task → External System

SMTs are applied after reading from Kafka but before writing to the external system.

Configuration

SMTs are configured in the connector configuration using the transforms property:

{
  "name": "jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-",
    
    "transforms": "RenameField,MaskField",
    "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.RenameField.renames": "user_id:userId",
    "transforms.MaskField.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.MaskField.fields": "ssn"
  }
}

Execution Order: SMTs are applied left-to-right in the transforms list (RenameField → MaskField).

Built-in SMTs

Kafka Connect provides several built-in SMTs. Below are the most commonly used ones.

1. InsertField

Purpose: Add a field to the record (e.g., timestamp, topic name, partition).

Use Cases:

  • Add Kafka metadata to records for traceability.
  • Inject static values (e.g., environment name, source system).

Configuration:

{
  "transforms": "AddTimestamp",
  "transforms.AddTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.AddTimestamp.timestamp.field": "ingest_time"
}

Result: Adds a field ingest_time with the Kafka record's timestamp.

Available Fields:

  • timestamp.field: Kafka record timestamp.
  • static.field + static.value: Static value (e.g., "env": "production").
  • topic.field: Kafka topic name.
  • partition.field: Kafka partition number.
  • offset.field: Kafka offset.

2. ReplaceField

Purpose: Rename or exclude fields.

Use Cases:

  • Rename fields to match target schema conventions (e.g., user_id → userId).
  • Remove sensitive fields before writing to Kafka.

Configuration (Rename):

{
  "transforms": "RenameField",
  "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.RenameField.renames": "user_id:userId,order_id:orderId"
}

Result: user_id → userId, order_id → orderId.

Configuration (Exclude):

{
  "transforms": "ExcludeField",
  "transforms.ExcludeField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.ExcludeField.exclude": "ssn,credit_card"
}

Result: Fields ssn and credit_card are removed.

Configuration (Whitelist):

{
  "transforms": "IncludeField",
  "transforms.IncludeField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.IncludeField.whitelist": "id,name,email"
}

Result: Only id, name, and email are kept; all other fields are removed.

3. MaskField

Purpose: Replace field values with a null or default value (for masking sensitive data).

Use Cases:

  • Mask PII (Personally Identifiable Information) before writing to Kafka.

Configuration:

{
  "transforms": "MaskSensitive",
  "transforms.MaskSensitive.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.MaskSensitive.fields": "ssn,credit_card",
  "transforms.MaskSensitive.replacement": "***REDACTED***"
}

Result: ssn and credit_card are replaced with ***REDACTED***.

4. ValueToKey

Purpose: Extract a field from the record value and use it as the record key.

Use Cases:

  • Key records by a specific field (e.g., user ID) for partitioning or log compaction.

Configuration:

{
  "transforms": "ExtractKey",
  "transforms.ExtractKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.ExtractKey.fields": "user_id"
}

Result: The user_id field from the value is copied to the record key.

Before:

  • Key: null
  • Value: {"user_id": 123, "name": "Alice"}

After:

  • Key: {"user_id": 123}
  • Value: {"user_id": 123, "name": "Alice"}

5. ExtractField

Purpose: Replace the entire record value with a single field from the original value.

Use Cases:

  • Extract a nested field to become the top-level value.

Configuration:

{
  "transforms": "ExtractPayload",
  "transforms.ExtractPayload.type": "org.apache.kafka.connect.transforms.ExtractField$Value",
  "transforms.ExtractPayload.field": "payload"
}

Before:

{"metadata": {"timestamp": 123}, "payload": {"user_id": 1, "name": "Alice"}}

After:

{"user_id": 1, "name": "Alice"}

6. Cast

Purpose: Change the data type of a field.

Use Cases:

  • Convert strings to integers, timestamps to epoch millis, etc.

Configuration:

{
  "transforms": "CastTypes",
  "transforms.CastTypes.type": "org.apache.kafka.connect.transforms.Cast$Value",
  "transforms.CastTypes.spec": "user_id:int32,age:int32,created_at:Timestamp"
}

Result: user_id and age are cast to int32, created_at is cast to Timestamp.

Supported Types:

  • int8, int16, int32, int64
  • float32, float64
  • boolean
  • string
  • Timestamp, Date, Time

7. TimestampConverter

Purpose: Convert timestamp fields between different formats.

Use Cases:

  • Convert Unix epoch (milliseconds) to ISO-8601 string.
  • Convert ISO-8601 string to Unix epoch.

Configuration (Epoch to String):

{
  "transforms": "ConvertTimestamp",
  "transforms.ConvertTimestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.ConvertTimestamp.field": "created_at",
  "transforms.ConvertTimestamp.target.type": "string",
  "transforms.ConvertTimestamp.format": "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
}

Result: created_at is converted from epoch millis (e.g., 1698412800000) to 2023-10-27T10:00:00.000Z.

Configuration (String to Epoch):

{
  "transforms": "ConvertTimestamp",
  "transforms.ConvertTimestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.ConvertTimestamp.field": "created_at",
  "transforms.ConvertTimestamp.target.type": "Timestamp",
  "transforms.ConvertTimestamp.format": "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
}

8. RegexRouter

Purpose: Modify the destination topic name based on a regex pattern.

Use Cases:

  • Route records to different topics based on source table names.
  • Apply topic prefixes/suffixes dynamically.

Configuration:

{
  "transforms": "RouteToEnv",
  "transforms.RouteToEnv.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.RouteToEnv.regex": "(.*)",
  "transforms.RouteToEnv.replacement": "prod-$1"
}

Result: Original topic orders → prod-orders.

9. Flatten

Purpose: Flatten nested structures into a flat structure.

Use Cases:

  • Simplify nested JSON for downstream systems that don't support nesting.

Configuration:

{
  "transforms": "FlattenValue",
  "transforms.FlattenValue.type": "org.apache.kafka.connect.transforms.Flatten$Value",
  "transforms.FlattenValue.delimiter": "_"
}

Before:

{"user": {"id": 1, "name": "Alice"}, "order": {"id": 100}}

After:

{"user_id": 1, "user_name": "Alice", "order_id": 100}

Chaining Multiple SMTs

Example: Combine Rename, Mask, and Insert

Scenario: Ingest user data from a database, rename fields, mask SSN, and add ingest timestamp.

Configuration:

{
  "transforms": "RenameField,MaskSSN,AddTimestamp",
  
  "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.RenameField.renames": "user_id:userId,first_name:firstName",
  
  "transforms.MaskSSN.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.MaskSSN.fields": "ssn",
  
  "transforms.AddTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.AddTimestamp.timestamp.field": "ingested_at"
}

Execution Order:

  1. RenameField: user_id → userId, first_name → firstName.
  2. MaskSSN: ssn → null.
  3. AddTimestamp: Add ingested_at with Kafka record timestamp.

Before:

{"user_id": 1, "first_name": "Alice", "ssn": "123-45-6789"}

After:

{"userId": 1, "firstName": "Alice", "ssn": null, "ingested_at": 1698412800000}

Key vs. Value Transformations

SMTs Apply to Keys or Values

Most SMTs have two variants:

  • $Key: Applies to the record key.
  • $Value: Applies to the record value.

Example: Rename fields in the value:

"transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value"

Example: Rename fields in the key:

"transforms.RenameKeyField.type": "org.apache.kafka.connect.transforms.ReplaceField$Key"

Independent Transformation

Keys and values can be transformed independently:

{
  "transforms": "RenameKeyField,RenameValueField",
  
  "transforms.RenameKeyField.type": "org.apache.kafka.connect.transforms.ReplaceField$Key",
  "transforms.RenameKeyField.renames": "id:userId",
  
  "transforms.RenameValueField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.RenameValueField.renames": "order_id:orderId"
}

Common Real-World SMT Patterns

Pattern 1: PII Masking Before Kafka

Use Case: Mask sensitive fields (SSN, credit card) before writing to Kafka.

{
  "transforms": "MaskPII",
  "transforms.MaskPII.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.MaskPII.fields": "ssn,credit_card"
}

Pattern 2: Key Extraction for Log Compaction

Use Case: Extract user_id from the value to use as the key, enabling log compaction.

{
  "transforms": "ExtractKey,SetKeySchema",
  
  "transforms.ExtractKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.ExtractKey.fields": "user_id",
  
  "transforms.SetKeySchema.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Key",
  "transforms.SetKeySchema.schema.name": "com.example.UserKey"
}

Pattern 3: Topic Routing by Environment

Use Case: Route records to environment-specific topics (e.g., dev-orders, prod-orders).

{
  "transforms": "AddEnvPrefix",
  "transforms.AddEnvPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.AddEnvPrefix.regex": "(.*)",
  "transforms.AddEnvPrefix.replacement": "prod-$1"
}

Pattern 4: Flatten Nested JSON for JDBC Sink

Use Case: Flatten nested JSON before writing to a relational database.

{
  "transforms": "FlattenValue",
  "transforms.FlattenValue.type": "org.apache.kafka.connect.transforms.Flatten$Value",
  "transforms.FlattenValue.delimiter": "_"
}

Pattern 5: Add Metadata for Observability

Use Case: Add Kafka metadata (topic, partition, offset, timestamp) to records for debugging.

{
  "transforms": "AddMetadata",
  "transforms.AddMetadata.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.AddMetadata.topic.field": "kafka_topic",
  "transforms.AddMetadata.partition.field": "kafka_partition",
  "transforms.AddMetadata.offset.field": "kafka_offset",
  "transforms.AddMetadata.timestamp.field": "kafka_timestamp"
}

Custom SMTs

When to Write a Custom SMT

Built-in SMTs cover most use cases, but you may need a custom SMT for:

  • Complex field manipulations (e.g., parsing custom date formats).
  • Business-specific transformations (e.g., calculating derived fields).

Implementing a Custom SMT

Custom SMTs implement the org.apache.kafka.connect.transforms.Transformation interface.

Example: Custom SMT to convert fields to uppercase.

public class ToUpperCase<R extends ConnectRecord<R>> implements Transformation<R> {
    private String fieldName;

    @Override
    public void configure(Map<String, ?> configs) {
        fieldName = (String) configs.get("field");
    }

    @Override
    public R apply(R record) {
        Struct value = (Struct) record.value();
        String originalValue = value.getString(fieldName);
        value.put(fieldName, originalValue.toUpperCase());
        return record.newRecord(record.topic(), record.kafkaPartition(), record.keySchema(),
                record.key(), value.schema(), value, record.timestamp());
    }

    @Override
    public ConfigDef config() {
        return new ConfigDef()
            .define("field", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Field to uppercase");
    }

    @Override
    public void close() {}
}

Configuration:

{
  "transforms": "UppercaseName",
  "transforms.UppercaseName.type": "com.example.ToUpperCase$Value",
  "transforms.UppercaseName.field": "name"
}

SMT Performance Considerations

1. SMTs Add Latency

Each SMT adds overhead to record processing. For high-throughput connectors, minimize SMT count.

2. Stateless = Fast

SMTs are stateless, so they scale linearly with record count. No memory overhead for windowing or joins.

3. Avoid Complex SMTs

Complex transformations (e.g., external API calls) should not be done in SMTs. Use Kafka Streams or custom code instead.

4. Test Throughput Impact

Benchmark connector throughput with and without SMTs to measure impact.

Key Takeaways

  1. SMTs are lightweight, record-level transformations applied inline during data ingestion/egress.
  2. Use SMTs for simple, stateless transformations (rename, mask, extract, cast).
  3. Built-in SMTs cover most use cases (InsertField, ReplaceField, MaskField, ValueToKey, etc.).
  4. Chaining allows multiple SMTs to be applied in sequence (left-to-right order).
  5. Key vs. Value: SMTs can transform keys ($Key) or values ($Value) independently.
  6. Custom SMTs can be implemented for business-specific transformations.
  7. Performance: SMTs add latency; avoid complex or stateful operations.

What's Next?

  • Module 5.1 covers performance tuning and throughput optimization.
  • Module 7 teaches custom connector development, including custom SMTs.
  • Module 6.1 explains error handling and retries (where SMT errors are handled).

Further Reading

End of Module 4.1