feat(connector): Add mqtt connector

.typos.toml

@@ -9,6 +9,7 @@ steam = "stream"        # You played with Steam games too much.
 # Some weird short variable names
 ot = "ot"
 bui = "bui"
+mosquitto = "mosquitto" # This is a MQTT broker.
 
 [default.extend-identifiers]
 

ci/scripts/gen-integration-test-yaml.py

@@ -34,7 +34,8 @@
     'big-query-sink': ['json'],
     'mindsdb': ['json'],
     'vector': ['json'],
-    'nats': ['json', 'protobuf'],
+    'nats': ['json'],
+    'mqtt': ['json'],
     'doris-sink': ['json'],
     'starrocks-sink': ['json'],
     'deltalake-sink': ['json'],

integration_tests/mqtt/create_sink.sql

@@ -0,0 +1,28 @@
+CREATE SINK mqtt_sink
+FROM
+  personnel
+WITH
+(
+    connector='mqtt',
+    url='tcp://mqtt-server',
+    topic= 'test',
+    type = 'append-only',
+    retain = 'true',
+    qos = 'at_least_once',
+) FORMAT PLAIN ENCODE JSON (
+    force_append_only='true',
+);
+
+INSERT INTO
+  personnel
+VALUES
+  (1, 'Alice'),
+  (2, 'Bob'),
+  (3, 'Tom'),
+  (4, 'Jerry'),
+  (5, 'Araminta'),
+  (6, 'Clover'),
+  (7, 'Posey'),
+  (8, 'Waverly');
+
+FLUSH;

integration_tests/mqtt/create_source.sql

@@ -0,0 +1,14 @@
+CREATE TABLE
+  personnel (id integer, name varchar);
+
+CREATE TABLE mqtt_source_table
+(
+  id integer,
+  name varchar,
+)
+WITH (
+    connector='mqtt',
+    url='tcp://mqtt-server',
+    topic= 'test',
+    qos = 'at_least_once',
+) FORMAT PLAIN ENCODE JSON;

integration_tests/mqtt/data_check

@@ -0,0 +1 @@
+personnel,mqtt_source_table

integration_tests/mqtt/docker-compose.yml

@@ -0,0 +1,49 @@
+---
+version: "3"
+services:
+  risingwave-standalone:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: risingwave-standalone
+  mqtt-server:
+    image: eclipse-mosquitto
+    command:
+    - sh
+    - -c
+    - echo "running command"; printf 'allow_anonymous true\nlistener 1883 0.0.0.0' > /mosquitto/config/mosquitto.conf; echo "starting service..."; cat /mosquitto/config/mosquitto.conf;/docker-entrypoint.sh;/usr/sbin/mosquitto -c /mosquitto/config/mosquitto.conf
+    ports:
+      - 1883:1883
+  etcd-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: etcd-0
+  grafana-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: grafana-0
+  minio-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: minio-0
+  prometheus-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: prometheus-0
+  message_queue:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: message_queue
+volumes:
+  compute-node-0:
+    external: false
+  etcd-0:
+    external: false
+  grafana-0:
+    external: false
+  minio-0:
+    external: false
+  prometheus-0:
+    external: false
+  message_queue:
+    external: false
+name: risingwave-compose

integration_tests/mqtt/query.sql

@@ -0,0 +1,8 @@
+select
+    *
+from
+    mqtt_source_table
+order by
+    id
+LIMIT
+    10;

integration_tests/mqtt/sink_check.py

@@ -0,0 +1,14 @@
+import sys
+import subprocess
+
+
+output = subprocess.Popen(["docker", "compose", "exec", "mqtt-server", "mosquitto_sub", "-h", "localhost", "-t", "test", "-p", "1883", "-C",  "1", "-W", "120"],
+                            stdout=subprocess.PIPE)
+rows = subprocess.check_output(["wc", "-l"], stdin=output.stdout)
+output.stdout.close()
+output.wait()
+rows = int(rows.decode('utf8').strip())
+print(f"{rows} rows in 'test'")
+if rows < 1:
+    print(f"Data check failed for case 'test'")
+    sys.exit(1)

src/connector/Cargo.toml

@@ -117,7 +117,10 @@ risingwave_common = { workspace = true }
 risingwave_jni_core = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
+rumqttc = { version = "0.22.0", features = ["url"] }
 rust_decimal = "1"
+rustls-native-certs =  "0.6"
+rustls-pemfile = "1"
 rw_futures_util = { workspace = true }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_derive = "1"
@@ -141,6 +144,7 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
 ] }
 tokio-postgres = { version = "0.7", features = ["with-uuid-1"] }
 tokio-retry = "0.3"
+tokio-rustls = "0.24"
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["codec", "io"] }
 tonic = { workspace = true }

src/connector/src/common.rs

@@ -684,3 +684,37 @@ impl NatsCommon {
         Ok(creds)
     }
 }
+
+pub(crate) fn load_certs(
+    certificates: &str,
+) -> ConnectorResult<Vec<tokio_rustls::rustls::Certificate>> {
+    let cert_bytes = if let Some(path) = certificates.strip_prefix("fs://") {
+        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
+    } else {
+        certificates.as_bytes().to_owned()
+    };
+
+    let certs = rustls_pemfile::certs(&mut cert_bytes.as_slice())?;
+
+    Ok(certs
+        .into_iter()
+        .map(tokio_rustls::rustls::Certificate)
+        .collect())
+}
+
+pub(crate) fn load_private_key(
+    certificate: &str,
+) -> ConnectorResult<tokio_rustls::rustls::PrivateKey> {
+    let cert_bytes = if let Some(path) = certificate.strip_prefix("fs://") {
+        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
+    } else {
+        certificate.as_bytes().to_owned()
+    };
+
+    let certs = rustls_pemfile::pkcs8_private_keys(&mut cert_bytes.as_slice())?;
+    let cert = certs
+        .into_iter()
+        .next()
+        .ok_or_else(|| anyhow!("No private key found"))?;
+    Ok(tokio_rustls::rustls::PrivateKey(cert))
+}

src/connector/src/error.rs

@@ -58,6 +58,9 @@ def_anyhow_newtype! {
     redis::RedisError => "Redis error",
     arrow_schema::ArrowError => "Arrow error",
     google_cloud_pubsub::client::google_cloud_auth::error::Error => "Google Cloud error",
+    tokio_rustls::rustls::Error => "TLS error",
+    rumqttc::v5::ClientError => "MQTT error",
+    rumqttc::v5::OptionError => "MQTT error",
 }
 
 pub type ConnectorResult<T, E = ConnectorError> = std::result::Result<T, E>;

src/connector/src/lib.rs

@@ -52,6 +52,7 @@ pub mod sink;
 pub mod source;
 
 pub mod common;
+pub mod mqtt_common;
 
 pub use paste::paste;
 

src/connector/src/macros.rs

@@ -32,6 +32,7 @@ macro_rules! for_all_classified_sources {
                 { Nexmark, $crate::source::nexmark::NexmarkProperties, $crate::source::nexmark::NexmarkSplit },
                 { Datagen, $crate::source::datagen::DatagenProperties, $crate::source::datagen::DatagenSplit },
                 { GooglePubsub, $crate::source::google_pubsub::PubsubProperties, $crate::source::google_pubsub::PubsubSplit },
+                { Mqtt, $crate::source::mqtt::MqttProperties, $crate::source::mqtt::split::MqttSplit },
                 { Nats, $crate::source::nats::NatsProperties, $crate::source::nats::split::NatsSplit },
                 { S3, $crate::source::filesystem::S3Properties, $crate::source::filesystem::FsSplit },
                 { Gcs, $crate::source::filesystem::opendal_source::GcsProperties , $crate::source::filesystem::OpendalFsSplit<$crate::source::filesystem::opendal_source::OpendalGcs> },