fix(error): fix compiling of lints (#16560)

Signed-off-by: Bugen Zhao <[email protected]>

ci/scripts/check-dylint.sh

@@ -6,6 +6,15 @@ set -euo pipefail
 source ci/scripts/common.sh
 unset RUSTC_WRAPPER # disable sccache, see https://github.com/mozilla/sccache/issues/861
 
+echo "--- List all available lints"
+output=$(cargo dylint list)
+if [ -z "$output" ]; then
+  echo "ERROR: No lints detected. There might be an issue with the configuration of the lints crate."
+  exit 1
+else
+  echo "$output"
+fi
+
 echo "--- Run dylint check (dev, all features)"
 # Instead of `-D warnings`, we only deny warnings from our own lints. This is because...
 # - Warnings from `check` or `clippy` are already checked in `check.sh`.

lints/Cargo.toml

@@ -16,7 +16,7 @@ path = "ui/format_error.rs"
 [dependencies]
 clippy_utils = { git = "https://github.com/rust-lang/rust-clippy", rev = "6fd0258e45105161b7e759a22e7350958e5cb0b1" }
 dylint_linting = "2.6.0"
-itertools = { workspace = true }
+itertools = "0.12"
 
 [dev-dependencies]
 dylint_testing = "2.6.0"

src/connector/src/sink/mod.rs

@@ -542,7 +542,11 @@ pub enum SinkError {
     #[error("Starrocks error: {0}")]
     Starrocks(String),
     #[error("Snowflake error: {0}")]
-    Snowflake(String),
+    Snowflake(
+        #[source]
+        #[backtrace]
+        anyhow::Error,
+    ),
     #[error("Pulsar error: {0}")]
     Pulsar(
         #[source]

src/connector/src/sink/snowflake.rs

@@ -16,7 +16,7 @@ use std::collections::HashMap;
 use std::fmt::Write;
 use std::sync::Arc;
 
-use anyhow::anyhow;
+use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use bytes::{Bytes, BytesMut};
 use risingwave_common::array::{Op, StreamChunk};
@@ -250,13 +250,13 @@ impl SnowflakeSinkWriter {
             .opendal_s3_engine
             .streaming_upload(&path)
             .await
-            .map_err(|err| {
-                SinkError::Snowflake(format!(
-                    "failed to create the streaming uploader of opendal s3 engine for epoch {}, error: {}",
-                    self.epoch,
-                    err
-                ))
-            })?;
+            .with_context(|| {
+                format!(
+                    "failed to create the streaming uploader of opendal s3 engine for epoch {}",
+                    self.epoch
+                )
+            })
+            .map_err(SinkError::Snowflake)?;
         Ok((uploader, file_suffix))
     }
 
@@ -276,12 +276,8 @@ impl SnowflakeSinkWriter {
         uploader
             .write_bytes(data)
             .await
-            .map_err(|err| {
-                SinkError::Snowflake(format!(
-                    "failed to write bytes when streaming uploading to s3 for snowflake sink, error: {}",
-                    err
-                ))
-            })?;
+            .context("failed to write bytes when streaming uploading to s3")
+            .map_err(SinkError::Snowflake)?;
         Ok(())
     }
 
@@ -293,12 +289,11 @@ impl SnowflakeSinkWriter {
             // there is no data to be uploaded for this epoch
             return Ok(None);
         };
-        uploader.finish().await.map_err(|err| {
-            SinkError::Snowflake(format!(
-                "failed to finish streaming upload to s3 for snowflake sink, error: {}",
-                err
-            ))
-        })?;
+        uploader
+            .finish()
+            .await
+            .context("failed to finish streaming upload to s3")
+            .map_err(SinkError::Snowflake)?;
         Ok(Some(file_suffix))
     }
 
@@ -315,12 +310,7 @@ impl SnowflakeSinkWriter {
                 "{}",
                 Value::Object(self.row_encoder.encode(row)?)
             )
-            .map_err(|err| {
-                SinkError::Snowflake(format!(
-                    "failed to write json object to `row_buf`, error: {}",
-                    err
-                ))
-            })?;
+            .unwrap(); // write to a `BytesMut` should never fail
         }
 
         // streaming upload in a chunk-by-chunk manner

src/connector/src/sink/snowflake_connector.rs

@@ -15,12 +15,12 @@
 use std::collections::HashMap;
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use anyhow::{anyhow, Context};
 use jsonwebtoken::{encode, Algorithm, EncodingKey, Header};
 use reqwest::{header, Client, RequestBuilder, StatusCode};
 use risingwave_common::config::ObjectStoreConfig;
 use risingwave_object_store::object::*;
 use serde::{Deserialize, Serialize};
-use thiserror_ext::AsReport;
 
 use super::doris_starrocks_connector::POOL_IDLE_TIMEOUT;
 use super::{Result, SinkError};
@@ -126,16 +126,12 @@ impl SnowflakeHttpClient {
         let jwt_token = encode(
             &header,
             &claims,
-            &EncodingKey::from_rsa_pem(self.private_key.as_ref()).map_err(|err| {
-                SinkError::Snowflake(format!(
-                    "failed to encode from provided rsa pem key, error: {}",
-                    err
-                ))
-            })?,
+            &EncodingKey::from_rsa_pem(self.private_key.as_ref())
+                .context("failed to encode from provided rsa pem key")
+                .map_err(SinkError::Snowflake)?,
         )
-        .map_err(|err| {
-            SinkError::Snowflake(format!("failed to encode jwt_token, error: {}", err))
-        })?;
+        .context("failed to encode jwt_token")
+        .map_err(SinkError::Snowflake)?;
         Ok(jwt_token)
     }
 
@@ -167,10 +163,10 @@ impl SnowflakeHttpClient {
         let response = builder
             .send()
             .await
-            .map_err(|err| SinkError::Snowflake(err.to_report_string()))?;
+            .map_err(|err| SinkError::Snowflake(anyhow!(err)))?;
 
         if response.status() != StatusCode::OK {
-            return Err(SinkError::Snowflake(format!(
+            return Err(SinkError::Snowflake(anyhow!(
                 "failed to make http request, error code: {}\ndetailed response: {:#?}",
                 response.status(),
                 response,
@@ -207,12 +203,8 @@ impl SnowflakeS3Client {
             &aws_secret_access_key,
             &aws_region,
         )
-        .map_err(|err| {
-            SinkError::Snowflake(format!(
-                "failed to create opendal s3 engine, error: {}",
-                err
-            ))
-        })?;
+        .context("failed to create opendal s3 engine")
+        .map_err(SinkError::Snowflake)?;
 
         Ok(Self {
             s3_bucket,

src/stream/src/executor/source/list_executor.rs

@@ -92,7 +92,7 @@ impl<S: StateStore> FsListExecutor<S> {
                         ]),
                     )),
                     Err(e) => {
-                        tracing::error!("Connector fail to list item: {e}");
+                        tracing::error!(error = %e.as_report(), "Connector fail to list item");
                         Err(e)
                     }
                 })