[feat][io] Add MQTT sink connector

README.md

@@ -58,6 +58,7 @@ mounting them into the `apachepulsar/pulsar` Docker image.
 | Kafka | Apache Kafka |
 | Kinesis | Amazon Kinesis Data Streams |
 | MongoDB | MongoDB |
+| MQTT | MQTT broker |
 | Redis | Redis |
 | Solr | Apache Solr |
 

distribution/io/build.gradle.kts

@@ -65,6 +65,7 @@ dependencies {
     connectorNars(project(":influxdb"))
     connectorNars(project(":redis"))
     connectorNars(project(":solr"))
+    connectorNars(project(":mqtt"))
     connectorNars(project(":dynamodb"))
     connectorNars(project(":alluxio"))
     connectorNars(project(":azure-data-explorer"))

gradle/libs.versions.toml

@@ -176,6 +176,7 @@ aerospike-client = "4.5.0"
 aws-sdk = "1.12.788"
 aws-sdk2 = "2.32.28"
 rabbitmq-client = "5.28.0"
+hivemq-mqtt-client = "1.3.13"
 cassandra-driver = "3.11.2"
 mongodb-driver = "5.4.0"
 influxdb-client = "7.3.0"
@@ -494,6 +495,7 @@ solr-test-framework = { module = "org.apache.solr:solr-test-framework", version.
 solr-core = { module = "org.apache.solr:solr-core", version.ref = "solr" }
 # Messaging
 rabbitmq-amqp-client = { module = "com.rabbitmq:amqp-client", version.ref = "rabbitmq-client" }
+hivemq-mqtt-client = { module = "com.hivemq:hivemq-mqtt-client", version.ref = "hivemq-mqtt-client" }
 nsq-j = { module = "com.sproutsocial:nsq-j", version.ref = "nsq-client" }
 # Time series
 influxdb-client-java = { module = "com.influxdb:influxdb-client-java", version.ref = "influxdb-client" }

mqtt/build.gradle.kts

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+plugins {
+    id("pulsar-connectors.java-conventions")
+    id("pulsar-connectors.nar-conventions")
+}
+
+dependencies {
+    implementation(libs.pulsar.io.common)
+    implementation(libs.pulsar.io.core)
+    implementation(libs.pulsar.functions.instance)
+    implementation(libs.jackson.databind)
+    implementation(libs.jackson.dataformat.yaml)
+    implementation(libs.commons.lang3)
+    implementation(libs.hivemq.mqtt.client)
+
+    testImplementation(libs.testcontainers)
+}

mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java

@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.mqtt;
+
+import com.hivemq.client.mqtt.MqttClient;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+@Connector(
+        name = "mqtt-sink",
+        type = IOType.SINK,
+        help = "A sink connector that moves messages from Pulsar to MQTT.",
+        configClass = MqttSinkConfig.class
+)
+@Slf4j
+public class MqttSink implements Sink<byte[]> {
+
+    private MqttSinkConfig mqttSinkConfig;
+    private Mqtt5AsyncClient mqttClient;
+    private MqttQos mqttQos;
+
+    @Override
+    public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
+        mqttSinkConfig = MqttSinkConfig.load(config, sinkContext);
+        mqttSinkConfig.validate();
+        mqttQos = MqttQos.fromCode(mqttSinkConfig.getQos());
+
+        var builder = MqttClient.builder()
+                .useMqttVersion5()
+                .serverHost(mqttSinkConfig.getServerHost())
+                .serverPort(mqttSinkConfig.getServerPort());
+
+        if (StringUtils.isNotBlank(mqttSinkConfig.getClientId())) {
+            builder = builder.identifier(mqttSinkConfig.getClientId());
+        }
+        if (mqttSinkConfig.isSslEnabled()) {
+            builder = builder.sslWithDefaultConfig();
+        }
+
+        mqttClient = builder.buildAsync();
+        if (StringUtils.isNotBlank(mqttSinkConfig.getUsername())) {
+            var authBuilder = mqttClient.connectWith()
+                    .cleanStart(mqttSinkConfig.isCleanStart())
+                    .keepAlive(mqttSinkConfig.getKeepAliveIntervalSec())
+                    .simpleAuth()
+                    .username(mqttSinkConfig.getUsername());
+            if (mqttSinkConfig.getPassword() != null) {
+                authBuilder = authBuilder.password(mqttSinkConfig.getPassword().getBytes(StandardCharsets.UTF_8));
+            }
+            authBuilder.applySimpleAuth()
+                    .send()
+                    .get(mqttSinkConfig.getConnectionTimeoutMs(), TimeUnit.MILLISECONDS);
+        } else {
+            mqttClient.connectWith()
+                    .cleanStart(mqttSinkConfig.isCleanStart())
+                    .keepAlive(mqttSinkConfig.getKeepAliveIntervalSec())
+                    .send()
+                    .get(mqttSinkConfig.getConnectionTimeoutMs(), TimeUnit.MILLISECONDS);
+        }
+        log.info("MQTT sink connected to {}:{}.",
+                mqttSinkConfig.getServerHost(), mqttSinkConfig.getServerPort());
+    }
+
+    @Override
+    public void write(Record<byte[]> record) {
+        try {
+            byte[] payload = record.getValue() == null ? new byte[0] : record.getValue();
+            mqttClient.publishWith()
+                    .topic(mqttSinkConfig.getTopic())
+                    .qos(mqttQos)
+                    .payload(payload)
+                    .send()
+                    .whenComplete((result, throwable) -> {
+                        if (throwable == null) {
+                            record.ack();
+                        } else {
+                            record.fail();
+                            log.warn("Failed to publish message to MQTT topic {}",
+                                    mqttSinkConfig.getTopic(), throwable);
+                        }
+                    });
+        } catch (Exception e) {
+            record.fail();
+            log.warn("Failed to schedule MQTT publish for topic {}", mqttSinkConfig.getTopic(), e);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (mqttClient == null) {
+            return;
+        }
+
+        try {
+            mqttClient.disconnectWith()
+                    .send()
+                    .get(mqttSinkConfig.getConnectionTimeoutMs(), TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            log.warn("Failed to disconnect MQTT client cleanly", e);
+        }
+    }
+}

mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSinkConfig.java

@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.mqtt;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.google.common.base.Preconditions;
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import lombok.Data;
+import lombok.experimental.Accessors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.io.common.IOConfigUtils;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+@Data
+@Accessors(chain = true)
+public class MqttSinkConfig implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "The MQTT broker host.")
+    private String serverHost;
+
+    @FieldDoc(
+            required = true,
+            defaultValue = "1883",
+            help = "The MQTT broker port.")
+    private int serverPort = 1883;
+
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "The MQTT topic to publish messages to.")
+    private String topic;
+
+    @FieldDoc(
+            defaultValue = "",
+            help = "MQTT client id used for the broker connection.")
+    private String clientId;
+
+    @FieldDoc(
+            defaultValue = "",
+            sensitive = true,
+            help = "MQTT username.")
+    private String username;
+
+    @FieldDoc(
+            defaultValue = "",
+            sensitive = true,
+            help = "MQTT password.")
+    private String password;
+
+    @FieldDoc(
+            defaultValue = "0",
+            help = "MQTT QoS level for outgoing messages. Valid values: 0, 1, 2.")
+    private int qos = 0;
+
+    @FieldDoc(
+            defaultValue = "60",
+            help = "MQTT keep alive interval in seconds.")
+    private int keepAliveIntervalSec = 60;
+
+    @FieldDoc(
+            defaultValue = "10000",
+            help = "Timeout in milliseconds for MQTT connect/disconnect operations.")
+    private long connectionTimeoutMs = 10000L;
+
+    @FieldDoc(
+            defaultValue = "true",
+            help = "Whether to start with a clean session.")
+    private boolean cleanStart = true;
+
+    @FieldDoc(
+            defaultValue = "false",
+            help = "Enable SSL/TLS with the client default SSL configuration.")
+    private boolean sslEnabled = false;
+
+    public static MqttSinkConfig load(String yamlFile) throws IOException {
+        ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+        return mapper.readValue(new File(yamlFile), MqttSinkConfig.class);
+    }
+
+    public static MqttSinkConfig load(Map<String, Object> map, SinkContext sinkContext) throws IOException {
+        return IOConfigUtils.loadWithSecrets(map, MqttSinkConfig.class, sinkContext);
+    }
+
+    public void validate() {
+        Preconditions.checkArgument(StringUtils.isNotBlank(serverHost), "serverHost cannot be blank");
+        Preconditions.checkArgument(serverPort > 0, "serverPort must be a positive integer");
+        Preconditions.checkArgument(StringUtils.isNotBlank(topic), "topic cannot be blank");
+        Preconditions.checkArgument(qos >= 0 && qos <= 2, "qos must be one of 0, 1, 2");
+        Preconditions.checkArgument(keepAliveIntervalSec >= 0, "keepAliveIntervalSec must be >= 0");
+        Preconditions.checkArgument(connectionTimeoutMs > 0, "connectionTimeoutMs must be > 0");
+    }
+}

mqtt/src/main/resources/META-INF/services/pulsar-io.yaml

@@ -0,0 +1,22 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+name: mqtt-sink
+description: MQTT sink connector
+sinkClass: org.apache.pulsar.io.mqtt.MqttSink
+sinkConfigClass: org.apache.pulsar.io.mqtt.MqttSinkConfig