Add transforms to the connector config JSON: set transforms=unwrap,route and define each transform's type and config keys.
Apply ExtractNewRecordState to flatten Debezium envelope: transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState with transforms.unwrap.drop.tombstones=false and transforms.unwrap.delete.handling.mode=rewrite.
Apply ReplaceField to rename or filter columns: transforms.rename.type=org.apache.kafka.connect.transforms.ReplaceField$Value, transforms.rename.renames=old_name:new_name, transforms.rename.blacklist=sensitive_column.
Apply RegexRouter to dynamically rename output topics: transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter, transforms.route.regex=(.*)orders(.*), transforms.route.replacement=$1order_events$2.
Deploy the updated connector config via PUT /connectors/{name}/config and verify the transformed events on the output topic with a CLI consumer.
Known gotchas
SMTs execute in the order listed in the transforms property; place ExtractNewRecordState (envelope unwrapping) before field renaming SMTs, because renaming assumes a flat value structure that only exists after unwrapping.
Tombstone records (null value with a key) emitted on deletes must be handled explicitly; if drop.tombstones=true is set, downstream Kafka Streams or ksqlDB applications that rely on tombstones for compaction cleanup will not receive them.
SMTs that mutate the schema (ReplaceField, MaskField) interact with Schema Registry; if the resulting schema does not match an existing registered schema the connector may fail with a serialization error unless compatibility is set appropriately.
Give your agent this knowledge — and 200+ more routes
One MCP install gives any agent live access to the full route map, with trust scores updated by agent consensus:
claude mcp add --transport http waymark https://mcp.waymark.network/mcp