Merge branch 'merge_master_3_11' into devy

.github/workflows/ci.yml

@@ -76,11 +76,17 @@ jobs:
           curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
       - name: Install Kafka
         run: |
-          wget --progress=dot --show-progress https://downloads.apache.org/kafka/3.7.0/kafka_2.12-3.7.0.tgz
+          wget --progress=dot --show-progress https://archive.apache.org/dist/kafka/3.5.0/kafka_2.12-3.5.0.tgz
           tar xvfz kafka*.tgz
           mkdir /tmp/kraft-combined-logs
           kafka_*/bin/kafka-storage.sh format -t 9v5PspiySuWU2l5NjTgRuA -c kafka_*/config/kraft/server.properties
           kafka_*/bin/kafka-server-start.sh -daemon kafka_*/config/kraft/server.properties
+      - name: Install mosquitto
+        run: |
+          sudo apt-get install -y mosquitto
+          sudo service mosquitto start
+      - name: Check Formatting
+        run: cargo fmt -- --check
       - name: Build
         run: cargo build --all-features
       - name: Validate API

crates/arroyo-api/src/connection_tables.rs

@@ -632,14 +632,18 @@ async fn get_schema(
         ));
     };
 
-    let resolver =
-        ConfluentSchemaRegistry::new(&endpoint, &table.topic, api_key.clone(), api_secret.clone())
-            .map_err(|e| {
-                bad_request(format!(
-                    "failed to fetch schemas from schema repository: {}",
-                    e
-                ))
-            })?;
+    let resolver = ConfluentSchemaRegistry::new(
+        &endpoint,
+        &table.subject(),
+        api_key.clone(),
+        api_secret.clone(),
+    )
+    .map_err(|e| {
+        bad_request(format!(
+            "failed to fetch schemas from schema repository: {}",
+            e
+        ))
+    })?;
 
     resolver.get_schema_for_version(None).await.map_err(|e| {
         bad_request(format!(

crates/arroyo-api/src/pipelines.rs

@@ -207,7 +207,7 @@ async fn try_register_confluent_schema(
     };
 
     let schema_registry =
-        ConfluentSchemaRegistry::new(&endpoint, &table.topic, api_key, api_secret)?;
+        ConfluentSchemaRegistry::new(&endpoint, &table.subject(), api_key, api_secret)?;
 
     match config.format.clone() {
         Some(Format::Avro(mut avro)) => {

crates/arroyo-connectors/Cargo.toml

@@ -48,7 +48,8 @@ regex = "1"
 ##########################
 
 # Kafka
-rdkafka = { version = "0.33", features = ["cmake-build"] }
+rdkafka = { version = "0.33", features = ["cmake-build", "tracing"] }
+rdkafka-sys = "4.5.0"
 
 # SSE
 eventsource-client = "0.12.0"
@@ -77,5 +78,11 @@ object_store = { workspace = true }
 deltalake = {version = "0.17", features = ["s3", "datafusion"] }
 async-compression = { version = "0.4.3", features = ["tokio", "zstd", "gzip"] }
 
+# MQTT
+rumqttc = { version = "0.23.0", features = ["url"] }
+rustls-native-certs =  "0.6"
+rustls-pemfile = "1"
+tokio-rustls = "0.24"
+
 [build-dependencies]
-glob = "0.3"
+glob = "0.3"