Learn Debezium - Custom SMT & Connector Extensions

#Introduction

The built-in transforms we covered in episode 6 handle many cases, but not all. Sometimes you need logic that isn't available — for example changing date formats, combining columns, or calling an external service for enrichment. Episode 17 covers writing custom Single Message Transforms (SMT) in Java.

An SMT is the lightest extension point for modifying events: it receives a SourceRecord, processes it, and returns a modified record or null. Because it runs inside the worker, an SMT doesn't require an additional runtime — just package it in a JAR and register it in the plugin path.

#Writing Custom Single Message Transforms

An SMT implements Kafka Connect's Transformation interface. Here's an example transform that adds a static environment field to the payload:

Custom SMT adding a field

public class AddEnvironment implements Transformation<SourceRecord> {
    private String env = "dev";
 
    @Override
    public SourceRecord apply(SourceRecord record) {
        Struct value = (Struct) record.value();
        value.put("environment", env);
        return record;
    }
 
    @Override
    public ConfigDef config() {
        return new ConfigDef()
            .define("environment", Type.STRING, "dev",
                    Importance.HIGH, "Environment name");
    }
 
    @Override
    public void close() {
    }
}

The class above adds an environment field to every payload. The value structure, which is a Struct, must be handled carefully — make sure the value schema already declares the field you're going to fill.

#Packaging and Deploying the Plugin

Package the code into a JAR with its dependencies, then put it in a plugin path the worker recognizes:

mvn -q clean package
cp target/add-environment-1.0.jar plugins/custom/

Mount the plugin folder into the Connect container and restart:

  connect:
    image: quay.io/debezium/connect:3.0
    volumes:
      - ./plugins:/kafka/connect/custom

Once the worker recognizes the plugin, register the SMT in the connector configuration:

{
  "transforms": "addEnv",
  "transforms.addEnv.type": "com.example.AddEnvironment",
  "transforms.addEnv.environment": "production"
}

Make sure transforms.addEnv.type uses the fully qualified class name that matches the package in the JAR.

#Use Cases: Masking, Enrichment, and Payload Normalization

Custom SMTs open up three groups of use cases:

• Advanced masking: disguising data based on patterns or cross-column correlation, going beyond the built-in MaskField.
• Enrichment: adding data from lookups, for example a country code from an IP address or a region from a branch code.
• Payload normalization: changing date formats, combining first_name and last_name columns, or dropping fields that aren't needed.

A simple enrichment example: before processing an event, the SMT calls a small in-memory lookup table to add region. Store the lookup in code or a static file so it doesn't add network latency on the streaming path.

#Maintaining Custom Connector Code

Custom code is technical debt that must be managed. Recommended practices:

• Unit tests: write a test for every transform with example input records.
• Versioning: follow semantic versioning; behavior changes are minor or major.
• Isolation: don't put all transforms in one giant JAR — separate them by domain.
• Documentation: document each transform's configuration and behavior for other teams.

mvn -q test

Every code change should go through the same pipeline as connector config: build, test, then deploy to staging before production.

#Interface Versus Schema

One common mistake when writing an SMT is changing the Struct structure without changing its schema. Kafka Connect validates records against the schema when sending them, so a structure change without the schema will trigger an error. If a transform adds or removes fields, build a new Schema with the matching Field objects.

For transforms that only read fields without changing them, return the same record unmodified. A transform that doesn't need to modify the payload is better written as a predicate rather than an SMT, so its overhead is minimal.

#Conclusion

Episode 17 gives you the ability to extend Debezium: writing custom SMTs in Java, packaging and deploying plugins to the worker, using them for masking, enrichment, and normalization, and maintaining the code with disciplined testing and versioning.

The key takeaways:

• An SMT implements the Transformation interface and modifies a SourceRecord.
• Plugins are registered through the plugin path and referenced by fully qualified class name.
• Custom SMTs excel at masking, enrichment, and normalization not found in the built-ins.
• Separate transforms by domain and don't combine them into one giant JAR.
• Write unit tests and run them in the pipeline before deploying to production.

In the next episode, episode 18, we'll discuss change event handling patterns — modeling insert, update, and delete downstream, handling out-of-order events and idempotent consumers, compaction, deduplication, and upsert, and building materialized views and CQRS.