MQTT

INSERT
INTO <mqtt-topic>
SELECT FIELD, ... 
FROM <kafka-topic>

-- Insert into /landoop/demo all fields from kafka_topicA
INSERT INTO /landoop/demo SELECT * FROM kafka_topicA

-- Insert into /landoop/demo all fields from dynamic field
INSERT INTO `$field` SELECT * FROM control.boxes.test WITHTARGET = field

export CONNECTOR=mqtt
docker-compose up -d mqtt

docker exec -ti mongo mongo

name=mqtt
connector.class=com.datamountaineer.streamreactor.connect.mqtt.sink.MqttSinkConnector
tasks.max=1
topics=orders
connect.mqtt.hosts=tcp://mqtt:1883
connect.mqtt.clean=true
connect.mqtt.timeout=1000
connect.mqtt.keep.alive=1000
connect.mqtt.service.quality=1
connect.mqtt.client.id=dm_sink_id
connect.mqtt.kcql=INSERT INTO /lenses/orders SELECT * FROM orders

docker exec -ti fastdata /bin/bash

connect-cli create mqtt < connector.properties

connect-cli create mqtt < connector.properties

connect-cli status mqtt

kafka-avro-console-producer \
 --broker-list localhost:9092 --topic orders \
 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"id","type":"int"},{"name":"created","type":"string"},{"name":"product","type":"string"},{"name":"price","type":"double"}, {"name":"qty", "type":"int"}]}'

{"id": 1, "created": "2016-05-06 13:53:00", "product": "OP-DAX-P-20150201-95.7", "price": 94.2, "qty":100}

docker exec -ti mqtt mosquitto_sub -t /lenses/orders

docker-compose down