Example: Configure MQTT Sink Connector for Eclipse Mosquitto Broker

The MQTT sink connector is used to publish records in an Apache Kafka® topic to MQTT brokers. This example demonstrates how to configure an MQTT sink connector for the Eclipse Mosquitto broker.

Set Up Mosquitto Broker and Clients

In this step, an Eclipse Mosquitto broker is set up by using Docker images.

  1. Install the Mosquitto utilities for your operating system.

  2. Create a config file named mosquitto.conf for the broker with the following contents.

    persistence false
    log_dest stdout
    allow_anonymous true
    connection_messages true
    

    Tip

    To start a broker that requires clients to authenticate using a username and password, change allow_anonymous to false and add password_file /etc/mosquitto/passwd. Use mosquitto_passwd -c password username to store passwords in this file.

  3. Start the Docker container.

    docker run --name mosquitto -p 1881:1881 -v `pwd`/mosquitto.conf:/mosquitto/config/mosquitto.conf eclipse-mosquitto
    
  4. Start the Docker container using this command.

    docker run --name mosquitto -p 1881:1881 -v `pwd`/mosquitto.conf:/mosquitto/config/mosquitto.conf -v `pwd`/password:/etc/mosquitto/passwd eclipse-mosquitto
    
  5. Verify that the broker is running by publishing a message to it.

    mosquitto_pub -h localhost -p 1881 -t my-mqtt-topic -m "sample-msg-1"
    

    Tip

    If a username and password are needed to connect to the Mosquitto broker, use the -u and -P arguments of the mosquitt_pub to pass in the correct credentials.

  6. To subscribe to all messages on a Mosquitto topic, use the mosquitto_sub command.

    mosquitto_sub -h localhost -p 1881 -t my-mqtt-topic
    

Install the MQTT Connector Plugin

Prerequisites
  1. Navigate to your Confluent Platform installation directory and run this command to install the latest version of the MQTT connector.

    confluent-hub install confluentinc/kafka-connect-mqtt:latest
    

    Tip

    To install a specific version of connector, replace latest with version number (for example, 1.1.0-preview) in the command above.

  2. Restart Connect to pick up the new plugin.

    confluent local stop connect && confluent local start connect
    
  3. Check if the MQTT plugin has been installed correctly and picked up by the plugin loader:

    curl -sS localhost:8083/connector-plugins | jq -c '.[] | select( .class | contains("Mqtt") )'
    

    This command should print the new MQTT connector plugins available to the worker.

    {"class": "io.confluent.connect.mqtt.MqttSinkConnector","type": "sink","version": "1.1.0-preview"}
    {"class": "io.confluent.connect.mqtt.MqttSourceConnector","type": "source","version": "1.1.0-preview"}
    

Configure Sink Connector

Prerequisites
  1. Create a configuration file named mqtt-sink-config.json with the following contents.

    {
        "name": "mqtt-sink",
        "config": {
            "connector.class": "io.confluent.connect.mqtt.MqttSinkConnector",
            "tasks.max": "1",
            "mqtt.server.uri": "tcp://127.0.0.1:1881",
            "topics":"kafka-topic-name",
            "mqtt.qos": "2",
            "mqtt.username": " Omit if MQTT broker supports anonymous mode ",
            "mqtt.password": " Omit if MQTT broker supports anonymous mode ",
            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
            "value.converter": "org.apache.kafka.connect.storage.StringConverter",
            "confluent.topic.bootstrap.servers": "kafka:9092",
            "confluent.license": " Omit to enable trial mode ",
            "confluent.topic.replication.factor": "1"
        }
    }
    
  2. Run this command to start the MQTT sink connector using the Confluent CLI confluent local load command.

    Caution

    You must include a double dash (--) between the topic name and your flag. For more information, see this post.

    confluent local load mqtt-sink -- -d mqtt-sink-config.json
    

    Your output should resemble:

    {
        "name": "mqtt-sink",
        "config": {
            "connector.class": "io.confluent.connect.mqtt.MqttSinkConnector",
            "tasks.max": "1",
            "mqtt.server.uri": "tcp://127.0.0.1:1881",
            "topics":"kafka-topic-name",
            "mqtt.qos": "2",
            "mqtt.username": " ##### ",
            "mqtt.password": " [hidden] ",
            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
            "value.converter": "org.apache.kafka.connect.storage.StringConverter",
            "confluent.topic.bootstrap.servers": "kafka:9092",
            "confluent.license": " [hidden] ",
            "confluent.topic.replication.factor": "1"
        }
        "tasks": [],
        "type": null
    }