How can Kafka mqtt connector send mqtt topic as key?

Question

I have a MQTT broker and a Kafka broker running, I have used the kafka-connector: https://github.com/Landoop/stream-reactor, with the next configuration:

name=Mqtt-Source
connector.class=com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector
tasks.max=1
connect.mqtt.kcql=INSERT INTO test SELECT * FROM + WITHCONVERTER=`com.datamountaineer.streamreactor.connect.converters.source.JsonSimpleConverter` WITHKEY(id)
connect.mqtt.connection.clean=true
connect.mqtt.connection.timeout=1000
connect.mqtt.connection.keep.alive=1000
connect.mqtt.client.id=test_mqtt_connector
connect.mqtt.converter.throw.on.error=true
connect.mqtt.hosts=tcp://mqtt-broker:1883
connect.mqtt.service.quality=1

In the kcql, I'm definning the field of the message that kafka should take as key, is it anyway to use the mqtt-topic as key? So I don't need to define the WITHKEY() in the kcql.

OneCricketeer OneCricketeer · Accepted Answer · 2018-11-19T17:03:19

I don't know about Landoop's KCQL, but assuming the topic is part of the message value, you can move it to the key like so

transforms=ReplaceKey,ExtractKey
transforms.ReplaceKey.type=org.apache.kafka.connect.transforms.ValueToKey
# change the field accordingly
transforms.ReplaceKey.fields=mqtt_topic
transforms.ExtractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
# make sure this is the same field as above
transforms.ExtractKey.field=mqtt_topic

If not, then you can staticly insert it

transforms=AddKey
transforms.AddKey.type=org.apache.kafka.connect.transforms.InsertField$Key
# The exclamation makes this a required field
transforms.AddKey.static.field=mqtt_topic!
transforms.AddKey.static.value="<<your topic name>>"

However, the above might not work with SELECT * FROM +, where you are selecting from all MQTT topics

How can Kafka mqtt connector send mqtt topic as key?

1 Answers