Single message transforms

Overview

Single message transforms, usually called SMTs, are small pieces of logic in Kafka Connect that change each record as it passes through a connector. They run one record at a time. They can rename a field, drop a column, mask a value, add a timestamp, change a topic name or flatten a nested structure. You configure them in the connector settings, with no custom code and no separate service.

The big idea is simple: keep the pipeline thin. SMTs are good for light, stateless, per-record edits. They are not meant for joins, aggregation, windows or lookups across records. For those jobs you use a stream processor such as Apache Flink or Spark Streaming. Used well, SMTs give you clean, safe, consistently shaped records before the data ever reaches a topic, a data lake or a model.

What it is

An SMT is a small class that implements a standard interface. It takes one Kafka Connect record in and returns one record out, or returns nothing, which drops the record. A connector can list several SMTs in order. That ordered list is called a transform chain.

SMTs can sit in two places:

• On a source connector, after the connector reads from the external system and before the record is written to Kafka.
• On a sink connector, after the record is read from Kafka and before the connector writes it to the external system.

Because the same mechanism works in both directions, a team can mask personal data on the way in, and a different team can reshape the same data for a warehouse on the way out.

What a record looks like to an SMT

A Kafka Connect record has these parts:

• A topic name.
• An optional partition number.
• A key and a key schema.
• A value and a value schema.
• A timestamp.
• Headers.

Most SMTs touch the value, the key, or the topic. Many SMTs come in pairs that end in $Key and $Value, so you pick which half of the record to change. For example ExtractField$Key works on the key and ExtractField$Value works on the value.

Built-in transforms

Apache Kafka ships a set of ready-made transforms. The list below covers the most common ones.

Many connector projects also publish their own transforms. A change data capture connector, for example, often provides one that unwraps the before and after envelope into a plain row. Treat such transforms as part of that connector project, and read its documentation for exact behaviour.

Where it came from

Kafka Connect was added to Apache Kafka to make it easy to move data in and out of Kafka without writing custom producers and consumers. Early on, users found that many pipelines needed small edits: rename a column, change a topic, drop a field. Putting that logic inside each connector would have meant every connector author writing the same code again.

The answer was a shared transformation API. It was introduced through a Kafka improvement proposal and shipped in a Kafka release after Connect itself. The goal was clear from the start. SMTs should be simple, stateless and per-record, so that Connect stays a data movement tool and does not become a second stream processing engine. Later versions added predicates, so a transform can apply only to records that match a condition, and header-focused transforms. The principle of one record in and at most one record out has stayed the same.

How it works

Step by step

1. A source connector reads data from an outside system, such as a database table or a file.
2. The connector turns each row or event into a Connect record, with a schema and a value.
3. The first SMT in the chain receives that record and returns a new record, or null to drop it.
4. The next SMT receives the output of the previous one. This repeats down the list.
5. The final record goes to a converter, which serialises it for Kafka. The converter may use a Schema Registry so the schema is registered and referenced by id.
6. For a sink connector, the order is reversed: the converter reads bytes from Kafka, the SMT chain runs, and the sink writes to the target.

An everyday analogy helps. Think of a mail room where every parcel goes down a short conveyor belt. At the first station a worker covers the home address with a sticker. At the second station another worker swaps the label for the right department. At the third, a worker removes parcels that are marked as samples. Each worker looks at one parcel only. Nobody holds a parcel back to compare it with others. That is how SMTs behave: simple stations, one item at a time, in a fixed order.

Configuration

SMTs are set with a few properties. You give each transform an alias, list the aliases in order, and then set the type and options for each alias. Here is a sample fragment for a sink connector.

{
  "name": "orders-sink",
  "config": {
    "connector.class": "com.example.SinkConnector",
    "topics": "orders",
    "transforms": "mask,drop,route",
    "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.mask.fields": "card_number,email",
    "transforms.drop.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.drop.exclude": "internal_notes",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "(.*)",
    "transforms.route.replacement": "$1_clean"
  }
}

The order in the transforms property is the order of execution. Here the mask runs first, then the drop, then the route.

Predicates

A predicate is a small test that decides whether a transform should run for a record. Built-in predicates include TopicNameMatches, HasHeaderKey and RecordIsTombstone. You attach a predicate to a transform and can negate it.

{
  "transforms": "dropTestRows",
  "transforms.dropTestRows.type": "org.apache.kafka.connect.transforms.Filter",
  "transforms.dropTestRows.predicate": "isTest",
  "predicates": "isTest",
  "predicates.isTest.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
  "predicates.isTest.pattern": "test-.*"
}

Schema handling

If a record has a schema, a transform that changes the structure must also change the schema. Built-in transforms do this for you and cache the updated schema so the work is not repeated for every record. Schemaless records (plain maps) are also supported by many transforms, though with fewer type guarantees.

Writing a custom SMT

When the built-in set is not enough, you can write your own. You implement the Transformation interface, provide a config definition, and package it as a plugin on the Connect worker path. The core method is apply, which takes a record and returns a record.

public class UpperCaseTopic<R extends ConnectRecord<R>> implements Transformation<R> {
  public R apply(R record) {
    return record.newRecord(
        record.topic().toUpperCase(), record.kafkaPartition(),
        record.keySchema(), record.key(),
        record.valueSchema(), record.value(),
        record.timestamp());
  }
  public ConfigDef config() { return new ConfigDef(); }
  public void configure(Map<String, ?> props) {}
  public void close() {}
}

Keep custom transforms free of network calls and shared mutable state. They run on the hot path of every record.

Key concepts and terms

A worked example

Suppose a source connector reads a customers table and publishes rows to Kafka. The raw record value looks like this:

{
  "customer_id": 1042,
  "full_name": "Sample Person",
  "email": "person@example.internal",
  "card_number": "0000111122223333",
  "country": "NL",
  "created_at": 1735689600000,
  "internal_notes": "called twice"
}

The goals are these:

• Hide the card number.
• Remove the internal notes.
• Rename customer_id to id.
• Add a field that says where the record came from.
• Send the data to a topic called customers_clean.

The chain

{
  "transforms": "mask,rename,source,route",
  "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.mask.fields": "card_number",
  "transforms.rename.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.rename.exclude": "internal_notes",
  "transforms.rename.renames": "customer_id:id",
  "transforms.source.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.source.static.field": "origin",
  "transforms.source.static.value": "crm-export",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "customers",
  "transforms.route.replacement": "customers_clean"
}

The result

After the chain, the record value looks like this:

{
  "id": 1042,
  "full_name": "Sample Person",
  "email": "person@example.internal",
  "card_number": "",
  "country": "NL",
  "created_at": 1735689600000,
  "origin": "crm-export"
}

Notice three things. First, MaskField replaced the card number with an empty string, which is the null value for a string field. The type stays the same, so downstream schemas do not break. Second, the rename and exclusion ran in one transform. Third, the topic name changed, so consumers of customers_clean never see the raw table name.

What the example does not do

The email is still visible. MaskField by default replaces values with a type-specific empty value. If you need a hashed value, so that the same email always becomes the same token, you need a custom transform or a stream processor. Hashing is also better done with a salt that is managed outside the connector config. This is a common gap, and the later sections return to it.

A tiny simulation in Python

The logic of a chain is easy to picture as plain functions. This snippet mirrors what the Connect worker does.

def mask(rec, fields):
    for f in fields:
        if f in rec:
            rec[f] = ""
    return rec

def rename(rec, mapping, drop):
    out = {}
    for k, v in rec.items():
        if k in drop:
            continue
        out[mapping.get(k, k)] = v
    return out

def insert(rec, name, value):
    rec[name] = value
    return rec

chain = [
    lambda r: mask(r, ["card_number"]),
    lambda r: rename(r, {"customer_id": "id"}, {"internal_notes"}),
    lambda r: insert(r, "origin", "crm-export"),
]

def run(record):
    for step in chain:
        record = step(record)
        if record is None:
            return None
    return record

The point of the sketch is the shape: each step takes one record, returns one record, and does not remember anything.

A second example: a sink that needs a flat table

Now take the opposite direction. A sink connector reads order events from a topic and writes to a table in a warehouse. The events are nested:

{
  "order": {"id": 7, "total": 59.9, "currency": "EUR"},
  "customer": {"id": 1042, "segment": "retail"},
  "meta": {"trace": "abc", "ingested_at": 1735689600000}
}

The table wants flat columns with simple names and no trace field. A chain for this is:

{
  "transforms": "flat,drop,ts",
  "transforms.flat.type": "org.apache.kafka.connect.transforms.Flatten$Value",
  "transforms.flat.delimiter": "_",
  "transforms.drop.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.drop.exclude": "meta_trace",
  "transforms.ts.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.ts.field": "meta_ingested_at",
  "transforms.ts.target.type": "Timestamp"
}

The output row has columns order_id, order_total, order_currency, customer_id, customer_segment and meta_ingested_at. The delimiter choice matters. If a later step in the warehouse expects dots or camel case, pick it here once, and keep it stable.

A checklist for reading a chain

When you review a connector config, ask these questions in order:

1. Which fields does each transform name, and do they exist at that point in the chain?
2. Does each transform act on the key or on the value?
3. Does any transform change a type that a registered schema depends on?
4. Is anything sensitive still present after the last step?
5. What happens to a null value and to a tombstone?
6. What is the final topic name or target table, and does it match the consumer's expectation?

A chain that passes all six questions is usually safe to promote.

How to measure the cost of a chain

Each transform adds a small amount of CPU per record. Most are cheap, but flattening large structures and converting schemas can add up at high volume. A simple method is to run the same connector with and without the chain on a copy of a topic, and compare task throughput and worker CPU. If the chain is a large share of the cost, move heavy reshaping to a batch job or a stream processor that can scale on its own.

Strengths and limitations

Strengths

• No extra infrastructure. The logic lives in the connector config and runs inside the Connect worker.
• Declarative and readable. A reviewer can read the chain and understand it.
• Reusable across connectors. The same mask can protect many pipelines.
• Early protection. Sensitive fields can be masked before data lands in Kafka, so no downstream consumer ever sees them.
• Low latency. Per-record work with no state is fast.
• Works in both directions, on sources and sinks.

Limitations

• Stateless only. You cannot count, join, deduplicate or window.
• One record in, at most one out. You cannot split a record into several without a custom approach or a different tool.
• No external lookups by design. Calling a service from a transform slows every record and is fragile.
• Debugging is harder than debugging code. A wrong class name or a typo in a field list often shows up only at runtime.
• Order matters, and a mistake in the order can silently change results. For example, renaming a field before masking it means the mask no longer finds it.
• Error handling is limited. A bad record can fail the task unless error tolerance and a dead letter queue are set up.
• Schema changes upstream can break a chain that names specific fields.
• Masking is not the same as strong anonymisation. Replacing a value with an empty string removes it, but a field combination can still identify a person.

How it compares with alternatives

A good rule: if you can describe the change in one sentence about a single record, try an SMT. If the sentence needs the words "all", "last", "per hour" or "look up", use a stream processor.

Role in tabular foundation models

Tabular foundation models learn from raw tables. They read rows and columns, handle mixed numeric and categorical types, cope with missing values, and often work without column names carrying meaning. They are pre-trained on large collections of synthetic or real tables, and at use time they learn from the table you give them through in-context learning. SMTs sit upstream of all this. They do not train or run a model. They shape the stream of rows that eventually becomes a table.

Here is how that shape matters in concrete terms.

Row and column structure

A model that reads a table wants a stable column order and stable column meaning. Streams from operational systems often contain nested structures, envelopes and extra metadata. A Flatten or ExtractField transform can turn a nested event into a flat row, which is what a table needs. A ReplaceField transform can drop columns that the model should not see, such as free text notes or internal flags.

Mixed types

Source systems often send numbers as strings, or dates as epoch integers. Cast and TimestampConverter help to make each column one clear type before it is stored. A model can handle mixed types, but a column that is a number in some rows and text in others adds noise. Cleaning at the connector keeps the stored table consistent.

Missing values

Tabular models are used to missing cells, and some are good at imputation. Still, there is a difference between a real missing value and a value that a transform erased. MaskField writes an empty or zero-like value that looks like real data. If a masked numeric column becomes zero, a model will treat zero as a true value, and that can distort learning or imputation. When masking a numeric field, prefer dropping the column or setting null where the schema allows it, so that the model sees the value as missing and not as a real zero.

No column names

Some tabular models can work even when column names are hidden or meaningless, because they learn from value patterns and relationships. Others use names as a hint. If you rename columns with ReplaceField, remember that a rename changes what a name-aware model sees. Keep names stable across the training extract and the scoring extract.

Pre-training data and synthetic tables

Pre-training corpora for tabular models mix synthetic and real tables. A pipeline that feeds a real table corpus should remove personal data first. SMTs are one clean place to do it: mask or drop identifiers at the edge, so the stored table never carries them. This also reduces the risk of a model memorising sensitive values. See /model/research for how tabular models are studied, and /model/architecture for how rows and columns are represented inside such a model.

In-context learning

With in-context learning, a tabular model is given labelled example rows and then asked to predict for new rows, without weight updates. The quality of those example rows is the quality of the prediction. An SMT chain that keeps the context rows clean, consistent and free of leakage helps directly. Leakage is worth a specific mention: a column that was added by InsertField, such as an ingestion offset or a processing timestamp, can accidentally predict the label. Drop such technical fields before building context tables.

Continual learning

When a model is refreshed over time, new data should arrive in the same shape as the old. An SMT chain pinned in configuration gives you that repeatability. A schema change at the source can be absorbed in the chain (for instance, by renaming a new column back to the old name) while the downstream tables stay the same. Pair this with schema drift detection and data contract validation so changes are noticed, not hidden.

Imputation and anomaly detection

For imputation and anomaly detection, the model needs to see the real spread of values. A transform that clips, hides or defaults values can hide anomalies. Do not use an SMT to repair values you want the model to judge. Use SMTs for structure, privacy and routing, and leave statistical judgement to the model. A Filter that removes test traffic is helpful; a filter that removes outliers is usually harmful to anomaly work.

Sector and semantic classification

Classification of columns or records into sectors or semantic types works best when source text is intact and consistently encoded. Keep descriptive fields, normalise their encoding, and avoid masking text that the classifier needs. If a field must be hidden for privacy, it may be better to remove the whole record from the training path than to leave a half-masked string.

Where a tabular model project would use SMTs in practice

• Edge masking and dropping of personal columns before data lands in topics.
• Routing different record types to different topics, so each topic maps to one table.
• Flattening change events into rows for a lake table in Apache Iceberg or Delta Lake.
• Adding provenance headers, so data lineage can trace a training table back to its source.
• Unwrapping a keyed event into a clean key and value for compaction.

Results of tabular models on public suites are summarised at /model/benchmarks. Those results depend on clean inputs, and good ingestion discipline is part of that.

Practical guidance and best practices

Design

• Keep chains short. Three or four transforms are easy to reason about. Ten are not.
• Put masking and dropping first, so that later transforms never touch sensitive data.
• Name aliases clearly (maskCard, dropNotes, not t1, t2).
• Prefer built-in transforms. Use custom code only when needed.
• Decide on one place for each concern: privacy at the edge, business logic in a stream processor, analytics in the warehouse.

Testing

• Test a chain on a sample topic before applying it to production.
• Keep a fixed set of sample records and their expected outputs, and run them whenever the chain or the Kafka version changes.
• Test a null value, a missing field, a tombstone and an unexpected type. These are the edge cases that cause failures.
• Check both the value and the schema of the output.

Operations

• Set error tolerance and a dead letter queue for sinks, so one bad record does not stop the task. Log the reason.
• Version control all connector configs.
• Watch task status and error counts. A rising error count after a source schema change is a signal.
• Use Schema Registry compatibility checks so that a transform that changes the shape of data cannot silently break consumers.
• When mirroring topics between clusters with MirrorMaker 2 or MSK Replicator, remember that SMTs applied before the write are part of the data that gets copied.

Security and privacy

• Treat masking as one control among several. Combine it with access control on topics and encryption in transit and at rest.
• If you need consistent tokens, use a keyed hash generated in a custom transform or a dedicated service, and keep the key outside connector configs.
• Review which fields are in the key. A sensitive key is copied into logs and into partition choices.

Common mistakes

• Ordering errors. Renaming a field before masking it, then masking by the old name, leaves the data exposed.
• Using a transform for stateful work, such as deduplication. It will not work, and it may seem to work in a small test.
• Forgetting the $Key or $Value suffix and changing the wrong half of the record.
• Masking a numeric field to zero and then feeding the table to a model that reads zero as real.
• Letting a custom transform call an external service on every record.
• Not planning for schema changes. A chain that names a column breaks when the column is renamed upstream.
• Leaving tombstones unhandled. A transform that expects a value fails on null.
• Dropping fields the downstream model needs, such as the timestamp used for time-based splits.
• Assuming SMTs run before the data is visible. In a sink, the raw record is already stored in Kafka. Masking at the sink protects the target, not the topic.
• Overlooking that headers can also carry sensitive data.

Frequently asked questions

What is a single message transform?

A single message transform is a small, stateless step in Kafka Connect that edits one record as it moves through a connector. It can rename or remove fields, mask values, change the topic, or filter records. You set it up in configuration, so no application code is needed for common tasks.

Can single message transforms join or aggregate records?

No. SMTs see one record at a time and keep no memory, so they cannot join streams, count events or build windows. For that work you should use a stream processor such as Apache Flink or Spark Streaming, or a Kafka Streams application.

Do SMTs run on source connectors or sink connectors?

Both. On a source connector they run before the record is written to Kafka, and on a sink connector they run after the record is read from Kafka and before it is written to the target. The same configuration style works for both directions.

Does the order of transforms matter?

Yes. The chain runs in the order listed, and each transform receives the output of the one before it. If you rename a field first and then try to mask it by its old name, the mask will not find it. Plan the order like a small recipe.

How do SMTs work with Schema Registry?

The transform changes the Connect record and its schema, and then the converter serialises it. If the converter uses a schema registry, the changed schema is registered and checked for compatibility at that point. A transform that changes field types can therefore trigger a compatibility failure, which is a useful early warning.

Is MaskField enough to anonymise personal data?

Not on its own. It replaces a value with an empty value of the same type, which hides that field, but other fields may still identify a person when combined. For strong de-identification, use hashing with managed keys, generalisation, or a tested de-identification job, and apply access controls as well.

Can I write my own SMT?

Yes. You implement the Transformation interface, define its config, and package it as a plugin on the Connect worker. Keep it fast and free of shared state, because it runs for every record. Test it with null values, missing fields and schema changes.

What happens when a transform fails on a record?

By default the task fails. For sink connectors you can configure error tolerance and a dead letter queue, so bad records are set aside with error details and the rest continue. Source connectors have fewer options, so test chains carefully before production use.

How do SMTs help with machine learning on tables?

They give a repeatable, early cleaning layer. You can flatten events into rows, drop leaky or sensitive columns, normalise types and route record types to separate topics. That produces consistent tables for training, in-context examples and scoring, with the sensitive material removed before storage.

When should I use a stream processor instead?

Use a stream processor when the change needs more than one record: joins, lookups, windows, counts, deduplication or splitting one event into many. If the rule can be said as an edit to a single record, an SMT is usually simpler and cheaper to run.

Glossary

• Alias: The short name given to a transform in connector config, used to set its options.
• Chain: The ordered list of transforms for one connector.
• Compaction: A Kafka feature that keeps the latest record per key in a topic.
• Connect worker: The process that runs connectors and their tasks.
• Converter: Serialises and deserialises records between Connect and bytes.
• Dead letter queue: A topic for records that could not be processed.
• Envelope: A struct that wraps before and after states of a row, plus metadata.
• Header: A key and value pair attached to a record outside the payload.
• Predicate: A condition that decides if a transform runs.
• Record: One message in Connect, with key, value, schemas, timestamp and headers.
• Schema: A description of the fields and types in a record.
• Stateless: Without memory of earlier records.
• Tombstone: A record with a null value that signals deletion.
• Transform: A class that changes a record or drops it.

Summary

Single message transforms are the lightweight editing layer of Kafka Connect. They act on one record at a time, are set up with configuration, and cover common needs such as masking, dropping, renaming, casting, flattening and routing. They are fast, easy to review, and reusable, but they cannot hold state, join data or call outside services.

For teams that feed tabular models, SMTs are a useful first line of hygiene. They shape events into flat, consistent rows, remove sensitive or leaky columns before storage, and keep the shape of data steady as a model is refreshed. Keep chains short, test them with edge cases, pair them with a schema registry and drift checks, and hand any stateful work to a proper stream processor. For related reading, see Kafka Connect, Apache Kafka, Schema Registry, and event-driven architecture.

Related

It belongs to Event streaming in my stack, under Data engineering and lakehouse. Others in the same group: