Kafka Connect InsertField transform with dynamic values

Question

I've run into a situation where I need to insert a new field to a Kafka Connect record, but it seems like the InsertField transform is limited to static values.

https://docs.confluent.io/current/connect/transforms/insertfield.html

Is there a way to add a dynamic value based off of other fields in the record?

The reason I need this is because I'm using JDBC Source and Sink Connectors to transfer data between two databases. On the sink side, I'm doing upserts. Because of that I need the following fields:

        "insert.mode": "upsert",
        "pk.mode": "record_value",
        "pk.fields": "TABLE_ID",

This works fine, but it ties me into having one connector file per table because in my source database all of the tables have primary keys in the form tableName_ID. So pk.fields is always going to be different.

I was thinking in my sink database I could add a new field that my application wouldn't know about, but Kafka Connect would use for the purposes of upsert primary keys. It would be called something like kafka_id and would be the same for every table. I'd like to add this field in my source config and then just update pk.fields in my sink. Is there any way I can do this? Do I need to write a custom transform? Thanks!

Answer 1

In your case the best option is your own Kafka Connect SMT: You can use topic info from ConnectRecord.

Example of SMT from How to Use Single Message Transforms in Kafka Connect article.