json on a topic that the JDBC connector publishes

We have the following json on a topic that the JDBC connector publishes to

{"APP_ID":3,"prod_ID":10,"APP_SETTING_NAME":"my_name","SETTING_KEY":"my_setting_key"}
Here's my connector file

name=data.app_setting
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
poll.interval.ms=500
tasks.max=4
mode=timestamp
query=SELECT APP_ID, APP_SETTING_NAME, SETTING_KEY,FROM MY_TABLE with (nolock)
timestamp.column.name=LAST_MOD_DATE
topic.prefix=data.app_setting

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
Now i want to insert a key to this message by multiplying the two integer fields - APP_SETTING_ID and USER_ID. So the key for this message becomes 9*10 = 90

Tagged:

Comments

  • You can try below steps

    query=SELECT APP_SETTING_ID, APP_SETTING_NAME, SETTING_KEY, (APP_SETTING_ID*USER_ID) as _key FROM MY_TABLE with (nolock)
    Then add an ExtractKey transform

    transforms=AddKeys,ExtractKey

    this make a map

    transforms.AddKeys.type=org.apache.kafka.connect.transforms.ValueToKey
    transforms.AddKeys.fields=_key

    this gets one field from the map

    transforms.ExtractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
    transforms.ExtractKey.field=_key

Sign In or Register to comment.