Argthtjtr

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.

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