fix format and clippy

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

src/connector/src/macros.rs

@@ -167,7 +167,7 @@ macro_rules! impl_split {
 
         $(
             impl TryFrom<SplitImpl> for $split {
-                type Error = crate::error::ConnectorError;
+                type Error = $crate::error::ConnectorError;
 
                 fn try_from(split: SplitImpl) -> std::result::Result<Self, Self::Error> {
                     match split {

src/connector/src/sink/log_store.rs

@@ -19,7 +19,6 @@ use std::future::{poll_fn, Future};
 use std::sync::Arc;
 use std::task::Poll;
 
-use anyhow::anyhow;
 use futures::{TryFuture, TryFutureExt};
 use risingwave_common::array::StreamChunk;
 use risingwave_common::bail;

src/connector/src/source/cdc/source/reader.rs

@@ -30,7 +30,7 @@ use risingwave_pb::connector_service::{
 use thiserror_ext::AsReport;
 use tokio::sync::mpsc;
 
-use crate::error::{ConnectorError, ConnectorResult as Result};
+use crate::error::{ConnectorError, ConnectorResult};
 use crate::parser::ParserConfig;
 use crate::source::base::SourceMessage;
 use crate::source::cdc::{CdcProperties, CdcSourceType, CdcSourceTypeTrait, DebeziumCdcSplit};
@@ -68,7 +68,7 @@ impl<T: CdcSourceTypeTrait> SplitReader for CdcSplitReader<T> {
         parser_config: ParserConfig,
         source_ctx: SourceContextRef,
         _columns: Option<Vec<Column>>,
-    ) -> Result<Self> {
+    ) -> ConnectorResult<Self> {
         assert_eq!(splits.len(), 1);
         let split = splits.into_iter().next().unwrap();
         let split_id = split.id();

src/connector/src/source/cdc/split.rs

@@ -14,7 +14,7 @@
 
 use std::marker::PhantomData;
 
-use anyhow::{anyhow, Context};
+use anyhow::Context;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/datagen/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/filesystem/file_common.rs

@@ -15,7 +15,6 @@ use std::fmt::Debug;
 use std::hash::Hash;
 use std::marker::PhantomData;
 
-use anyhow::anyhow;
 use aws_sdk_s3::types::Object;
 use risingwave_common::types::{JsonbVal, Timestamptz};
 use serde::{Deserialize, Serialize};

src/connector/src/source/filesystem/nd_streaming.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use bytes::BytesMut;
 use futures::io::Cursor;
 use futures::AsyncBufReadExt;

src/connector/src/source/google_pubsub/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/kafka/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/kinesis/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/nats/source/reader.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use async_nats::jetstream::consumer;
 use async_trait::async_trait;
 use futures::StreamExt;

src/connector/src/source/nexmark/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/pulsar/enumerator/client.rs

@@ -47,7 +47,7 @@ impl SplitEnumerator for PulsarSplitEnumerator {
     async fn new(
         properties: PulsarProperties,
         _context: SourceEnumeratorContextRef,
-    ) -> Result<PulsarSplitEnumerator> {
+    ) -> ConnectorResult<PulsarSplitEnumerator> {
         let pulsar = properties
             .common
             .build_client(&properties.oauth, &properties.aws_auth_props)

src/connector/src/source/pulsar/split.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
 use risingwave_common::types::JsonbVal;
 use serde::{Deserialize, Serialize};
 

src/connector/src/source/test_source.rs

@@ -15,7 +15,6 @@
 use std::collections::HashMap;
 use std::sync::{Arc, OnceLock};
 
-use anyhow::anyhow;
 use async_trait::async_trait;
 use parking_lot::Mutex;
 use risingwave_common::bail;

src/meta/src/controller/catalog.rs

@@ -662,7 +662,7 @@ impl CatalogController {
             let ret = src_manager.register_source(&pb_source).await;
             if let Err(e) = ret {
                 txn.rollback().await?;
-                return Err(e.into());
+                return Err(e);
             }
         }
         txn.commit().await?;

src/meta/src/rpc/ddl_controller.rs

@@ -432,7 +432,7 @@ impl DdlController {
                     mgr.catalog_manager
                         .cancel_create_source_procedure(&source)
                         .await?;
-                    return Err(e.into());
+                    return Err(e);
                 }
 
                 mgr.catalog_manager