Add a Single Message Transform to a Kafka Connect Sink: Drop a Field and Rename Another

What You’ll Achieve

In this guide you’ll learn how to configure a Single Message Transform (SMT) on a Kafka Connect sink connector so that a specific field is removed from each record and another field is renamed before the data lands in the downstream system. The result is a clean, target‑schema‑compatible payload without needing to modify the producer or the source topic.

Prerequisites

• A running Kafka Connect cluster (stand‑alone or distributed) with at least one sink connector in operation.
• Source topic containing JSON records (Avro or other formats work too, but the example below uses JSON).
• Downstream system (e.g., PostgreSQL, Elasticsearch, file sink) that expects the transformed schema.
• Access to the Connect REST API (default port 8083) and the ability to edit connector configs.

Step‑by‑Step Procedure

1. Identify the Fields

   Suppose your source record looks like:

   {"user_id": 123, "email": "[contact removed]", "password_hash": "abc123", "last_login": "2026-10-19T12:34:56Z"}

   You want to drop password_hash and rename last_login to login_time.

2. Prepare the SMT Configuration

   Add the following block to the connector’s JSON configuration. Replace connector_name and other placeholders with your actual values.

   {
     "name": "connector_name",
     "config": {
       "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
       "tasks.max": "1",
       "topics": "source_topic",
       "connection.url": "jdbc:postgresql://db:5432/mydb",
       "auto.create": "true",
       "transforms": "renameAndDrop",
       "transforms.renameAndDrop.type": "org.apache.kafka.connect.transforms.ReplaceField",
       "transforms.renameAndDrop.blacklist": "password_hash",
       "transforms.renameAndDrop.rename": "last_login:login_time",
       "errors.tolerance": "none",
       "errors.log.enable": "true"
     }
   }

   Key points:

   • transforms lists the names of SMTs in the order they run.
   • Each SMT name gets a dedicated transforms..type property.
   • For ReplaceField, use blacklist to drop fields and rename to map old to new names.
   • Field names are case‑sensitive; double‑check spelling.

3. Deploy or Update the Connector

   Use the Connect REST API to create or update the connector. If the connector already exists, use PUT to replace its config.

   curl -X PUT http://localhost:8083/connectors/connector_name/config \
        -H "Content-Type: application/json" \
        -d @connector_config.json
   

   Replace @connector_config.json with the path to the file containing the JSON above. The worker will restart the connector with the new SMT.

4. Validate the Transformation

   Produce a test record to the source topic:

   echo '{"user_id":123,"email":"[contact removed]","password_hash":"abc123","last_login":"2026-10-19T12:34:56Z"}' \
     | kafka-console-producer --broker-list localhost:9092 --topic source_topic --property value.serializer=org.apache.kafka.common.serialization.StringSerializer
   

   Consume the sink output (e.g., query PostgreSQL or tail the file sink) and confirm the record now contains login_time and no password_hash.

   Check the worker logs for DEBUG messages from org.apache.kafka.connect.transforms.ReplaceField to verify each record was processed. If you see errors, review the errors.log.enable output.

5. Recover from SMT Failures
   • If the SMT mis‑configures and records are skipped, set errors.tolerance=all to let the connector continue while logging failures.
   • To temporarily bypass the SMT, remove the transforms block and redeploy. This restores the original record structure.
   • Always monitor disk usage when errors.log.enable=true because frequent failures can fill logs quickly.

Performance and Compatibility Notes

• SMTs run in the connector worker thread; complex logic can increase CPU usage. For high‑volume topics, consider offloading heavy transforms to Kafka Streams.
• When using Avro with a Schema Registry, ReplaceField may produce a record that no longer matches the registered schema. Verify that the target system accepts the new schema or re‑register it.
• Schema evolution rules (e.g., adding a new field) can break downstream consumers if the SMT removes a field they rely on. Test thoroughly in a staging environment.

Practical Checklists

• Verify the connector’s transforms block via the REST API: curl http://localhost:8083/connectors/connector_name/config.
• Run kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group connector_name to confirm the connector is active and consuming.
• Check the target system’s schema after the SMT to ensure compatibility.

Conclusion

Adding a ReplaceField SMT to a Kafka Connect sink connector is a lightweight way to sanitize and adapt data on the fly. By carefully configuring the blacklist and rename maps, you can enforce downstream schema rules without touching the producer or source topic. Always test the SMT in a controlled environment and monitor logs for unexpected failures.