chore: Fix dylint errors

risingwavelabs · Mar 3, 2024 · 0d1861c · 0d1861c
1 parent 2788587
commit 0d1861c
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 8 deletions.
diff --git a/integration_tests/mqtt/create_source.sql b/integration_tests/mqtt/create_source.sql
@@ -9,7 +9,8 @@ CREATE TABLE mqtt_source_table
 WITH (
     connector='mqtt',
     host='mqtt-server',
-    topic= 'test'
+    topic= 'test',
+    qos = '1'
 ) FORMAT PLAIN ENCODE JSON;
 
 
@@ -23,7 +24,7 @@ WITH
     topic= 'test',
     type = 'append-only',
     force_append_only='true',
-    retain = 'true',
+    retain = 'false',
     qos = '1'
   );
 
@@ -39,5 +40,4 @@ VALUES
   (7, 'Posey'),
   (8, 'Waverly');
 
-
-FLUSH;
+FLUSH;
diff --git a/src/connector/src/sink/mqtt.rs b/src/connector/src/sink/mqtt.rs
@@ -23,6 +23,7 @@ use rumqttc::v5::mqttbytes::QoS;
 use rumqttc::v5::ConnectionError;
 use serde_derive::Deserialize;
 use serde_with::serde_as;
+use thiserror_ext::AsReport;
 use tokio_retry::strategy::{jitter, ExponentialBackoff};
 use tokio_retry::Retry;
 use with_options::WithOptions;
@@ -171,11 +172,11 @@ impl MqttSinkWriter {
 
                         if let ConnectionError::MqttState(rumqttc::v5::StateError::Io(err)) = err {
                             if err.kind() != std::io::ErrorKind::ConnectionAborted {
-                                tracing::error!("[Sink] Failed to poll mqtt eventloop: {}", err);
+                                tracing::error!("[Sink] Failed to poll mqtt eventloop: {}", err.as_report());
                                 std::thread::sleep(std::time::Duration::from_secs(1));
                             }
                         } else {
-                            tracing::error!("[Sink] Failed to poll mqtt eventloop: {}", err);
+                            tracing::error!("[Sink] Failed to poll mqtt eventloop: {}", err.as_report());
                             std::thread::sleep(std::time::Duration::from_secs(1));
                         }
                     }

diff --git a/src/connector/src/source/mqtt/enumerator/mod.rs b/src/connector/src/source/mqtt/enumerator/mod.rs
@@ -20,6 +20,7 @@ use async_trait::async_trait;
 use risingwave_common::bail;
 use rumqttc::v5::{ConnectionError, Event, Incoming};
 use rumqttc::Outgoing;
+use thiserror_ext::AsReport;
 use tokio::sync::RwLock;
 
 use super::source::MqttSplit;
@@ -93,7 +94,7 @@ impl SplitEnumerator for MqttSplitEnumerator {
                         tracing::error!(
                             "[Enumerator] Failed to subscribe to topic {}: {}",
                             topic,
-                            err
+                            err.as_report(),
                         );
                         connected_clone.store(false, std::sync::atomic::Ordering::Relaxed);
                         cloned_client

diff --git a/src/connector/src/source/mqtt/source/reader.rs b/src/connector/src/source/mqtt/source/reader.rs
@@ -18,6 +18,7 @@ use risingwave_common::bail;
 use rumqttc::v5::mqttbytes::v5::Filter;
 use rumqttc::v5::mqttbytes::QoS;
 use rumqttc::v5::{ConnectionError, Event, Incoming};
+use thiserror_ext::AsReport;
 
 use super::message::MqttMessage;
 use super::MqttSplit;
@@ -105,7 +106,7 @@ impl CommonSplitReader for MqttSplitReader {
                     if let ConnectionError::Timeout(_) = e {
                         continue;
                     }
-                    tracing::error!("[Reader] Failed to poll mqtt eventloop: {}", e);
+                    tracing::error!("[Reader] Failed to poll mqtt eventloop: {}", e.as_report());
                     client
                         .subscribe_many(
                             splits