[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,53 @@
+/*
+ * 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")
+}
+
+val integrationTest by sourceSets.creating {
+    compileClasspath += sourceSets.main.get().output + configurations.testCompileClasspath.get()
+    runtimeClasspath += output + sourceSets.main.get().output + configurations.testRuntimeClasspath.get()
+    resources.srcDir(rootProject.file("gradle/test-resources"))
+}
+
+configurations[integrationTest.implementationConfigurationName].extendsFrom(configurations.testImplementation.get())
+configurations[integrationTest.runtimeOnlyConfigurationName].extendsFrom(configurations.testRuntimeOnly.get())
+
+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.guava)
+    implementation(libs.hivemq.mqtt.client)
+
+    add(integrationTest.implementationConfigurationName, libs.testcontainers)
+}
+
+tasks.register<Test>("integrationTest") {
+    description = "Runs MQTT integration tests that require Docker."
+    group = LifecycleBasePlugin.VERIFICATION_GROUP
+    testClassesDirs = integrationTest.output.classesDirs
+    classpath = integrationTest.runtimeClasspath
+    shouldRunAfter("test")
+}

mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java

@@ -0,0 +1,150 @@
+/*
+ * 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 static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import com.hivemq.client.mqtt.MqttClient;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.SinkContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.DockerImageName;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class MqttSinkIntegrationTest {
+
+    private static final Logger log = LoggerFactory.getLogger(MqttSinkIntegrationTest.class);
+
+    private static final int MQTT_PORT = 1883;
+    private static final String TEST_TOPIC = "pulsar/mqtt/e2e";
+    private static final DockerImageName MOSQUITTO_IMAGE = DockerImageName.parse("eclipse-mosquitto:2");
+
+    private final GenericContainer<?> mqttContainer = new GenericContainer<>(MOSQUITTO_IMAGE)
+            .withExposedPorts(MQTT_PORT);
+
+    @BeforeClass(alwaysRun = true)
+    public void beforeClass() {
+        mqttContainer.start();
+    }
+
+    @AfterClass(alwaysRun = true)
+    public void afterClass() {
+        mqttContainer.stop();
+    }
+
+    @Test
+    public void testWriteE2EWithMosquitto() throws Exception {
+        BlockingQueue<String> receivedPayloads = new LinkedBlockingQueue<>();
+        CountDownLatch ackLatch = new CountDownLatch(3);
+        AtomicBoolean failCalled = new AtomicBoolean(false);
+
+        Mqtt5AsyncClient subscriber = MqttClient.builder()
+                .useMqttVersion5()
+                .serverHost(mqttContainer.getHost())
+                .serverPort(mqttContainer.getMappedPort(MQTT_PORT))
+                .identifier("mqtt-sink-e2e-subscriber")
+                .buildAsync();
+
+        subscriber.connectWith()
+                .cleanStart(true)
+                .send()
+                .get(10, TimeUnit.SECONDS);
+        subscriber.subscribeWith()
+                .topicFilter(TEST_TOPIC)
+                .callback(publish -> receivedPayloads.add(
+                        new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8)))
+                .send()
+                .get(10, TimeUnit.SECONDS);
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("serverHost", mqttContainer.getHost());
+        config.put("serverPort", mqttContainer.getMappedPort(MQTT_PORT));
+        config.put("topic", TEST_TOPIC);
+        config.put("qos", 1);
+        config.put("connectionTimeoutMs", 10000);
+        config.put("clientId", "mqtt-sink-e2e-publisher");
+
+        SinkContext sinkContext = mock(SinkContext.class);
+        try (MqttSink sink = new MqttSink()) {
+            sink.open(config, sinkContext);
+
+            for (int i = 0; i < 3; i++) {
+                sink.write(new TestRecord(("msg-" + i).getBytes(StandardCharsets.UTF_8), ackLatch, failCalled));
+            }
+
+            assertTrue(ackLatch.await(10, TimeUnit.SECONDS), "Timed out waiting for record.ack()");
+            assertFalse(failCalled.get(), "record.fail() should not be called on successful publish");
+
+            assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-0");
+            assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-1");
+            assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-2");
+        } finally {
+            try {
+                subscriber.disconnectWith()
+                        .sessionExpiryInterval(0)
+                        .send()
+                        .get(10, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                log.warn("Failed to disconnect MQTT subscriber in test cleanup", e);
+            }
+        }
+    }
+
+    private static final class TestRecord implements Record<byte[]> {
+        private final byte[] value;
+        private final CountDownLatch ackLatch;
+        private final AtomicBoolean failCalled;
+
+        private TestRecord(byte[] value, CountDownLatch ackLatch, AtomicBoolean failCalled) {
+            this.value = value;
+            this.ackLatch = ackLatch;
+            this.failCalled = failCalled;
+        }
+
+        @Override
+        public byte[] getValue() {
+            return value;
+        }
+
+        @Override
+        public void ack() {
+            ackLatch.countDown();
+        }
+
+        @Override
+        public void fail() {
+            failCalled.set(true);
+        }
+    }
+}

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

@@ -0,0 +1,136 @@
+/*
+ * 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 com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder;
+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",
+        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 = buildClient(builder);
+        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());
+    }
+
+    Mqtt5AsyncClient buildClient(Mqtt5ClientBuilder builder) {
+        return builder.buildAsync();
+    }
+
+    @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 (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("MQTT disconnect was interrupted", e);
+        } catch (Exception e) {
+            log.warn("Failed to disconnect MQTT client cleanly", e);
+        }
+    }
+}