feat(connectors): add MQTT sink connector

Cargo.toml

@@ -43,6 +43,7 @@ members = [
     "core/connectors/sinks/influxdb_sink",
     "core/connectors/sinks/meilisearch_sink",
     "core/connectors/sinks/mongodb_sink",
+    "core/connectors/sinks/mqtt_sink",
     "core/connectors/sinks/postgres_sink",
     "core/connectors/sinks/quickwit_sink",
     "core/connectors/sinks/rabbitmq_sink",

core/connectors/runtime/example_config/connectors/mqtt_sink.toml

@@ -0,0 +1,80 @@
+# 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.
+
+type = "sink"
+key = "mqtt"
+enabled = true
+version = 0
+name = "MQTT sink"
+path = "target/release/libiggy_connector_mqtt_sink"
+plugin_config_format = "json"
+
+[[streams]]
+stream = "iot"
+topics = ["telemetry"]
+schema = "raw"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "mqtt_sink_connector"
+
+[plugin_config]
+# Required broker address. Do not add credentials or MQTT options to this URL.
+broker_url = "mqtt://127.0.0.1:1883"
+protocol = "mqtt5"
+clean_start = true
+session_expiry_interval = 3600
+
+# Select one routing block. Keep one topic_mode line active.
+# Static routing publishes every Iggy record to one MQTT topic.
+topic = "iggy.messages"
+topic_mode = "static"
+qos = 1
+retain = false
+
+# Template routing creates an MQTT topic from the Iggy stream and topic.
+# To select it, comment the active static topic and topic_mode lines above.
+# topic_mode = "template"
+# topic_template = "{stream}/{topic}"
+
+# Dynamic routing reads mqtt.topic from Iggy headers or an MQTT 5 envelope.
+# It uses fallback_topic when the record has no valid MQTT topic.
+# To select it, comment the active static topic and topic_mode lines above.
+# topic_mode = "dynamic"
+# fallback_topic = "{stream}/{topic}"
+
+# Connection, publication, and retry settings.
+keep_alive = "30s"
+request_capacity = 32
+publish_timeout = "30s"
+shutdown_timeout = "30s"
+# Retries after the initial connection attempt. Each publish window receives
+# this many retries after its initial uncertain publication attempt.
+max_retries = 5
+retry_delay = "1s"
+max_retry_delay = "30s"
+max_in_flight = 10
+verbose_logging = false
+
+# Optional broker authentication.
+# username = "emqx-user"
+# password = "replace-me"
+
+# Optional TLS configuration for mqtts:// brokers.
+# [plugin_config.tls]
+# ca_file = "/etc/iggy/certs/emqx-ca.pem"
+# client_cert_file = "/etc/iggy/certs/client-cert.pem"
+# client_key_file = "/etc/iggy/certs/client-key.pem"

core/connectors/sinks/README.md

@@ -14,6 +14,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | **iceberg_sink** | Writes data to Apache Iceberg tables via REST catalog with S3/GCS/Azure storage |
 | **influxdb_sink** | Writes messages to InfluxDB as line-protocol points; supports both V2 (org/bucket, Flux) and V3 (db, SQL) |
 | **meilisearch_sink** | Indexes messages in Meilisearch for full-text search |
+| **mqtt_sink** | Publishes messages to MQTT brokers |
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
 | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics |
 | **redshift_sink** | Stores messages in Redshift warehouse tables with configurable schemas vis S3 as staging |

core/connectors/sinks/mqtt_sink/Cargo.toml

@@ -0,0 +1,56 @@
+# 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]
+name = "iggy_connector_mqtt_sink"
+version = "0.5.0"
+description = "Iggy MQTT sink connector."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "mqtt", "messaging", "streaming", "sink"]
+categories = ["command-line-utilities", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+bytes = { workspace = true }
+humantime = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+rumqttc-v4 = { workspace = true }
+rumqttc-v5 = { workspace = true }
+rustls-native-certs = { workspace = true }
+rustls-pemfile = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+url = { workspace = true }
+uuid = { workspace = true }
+
+[lints]
+workspace = true