fix: refine error prompting

src/common/src/error.rs

@@ -20,12 +20,10 @@ use std::io::Error as IoError;
 use std::sync::Arc;
 
 use memcomparable::Error as MemComparableError;
-use prost::Message;
 use risingwave_pb::common::Status;
 use risingwave_pb::ProstFieldNotFound;
 use thiserror::Error;
 use tokio::task::JoinError;
-use tonic::metadata::{MetadataMap, MetadataValue};
 use tonic::Code;
 
 use crate::array::ArrayError;
@@ -191,17 +189,8 @@ impl From<RwError> for tonic::Status {
             ErrorCode::OK => tonic::Status::ok(err.to_string()),
             ErrorCode::ExprError(e) => tonic::Status::invalid_argument(e.to_string()),
             ErrorCode::PermissionDenied(e) => tonic::Status::permission_denied(e),
-            _ => {
-                let bytes = {
-                    let status = err.to_status();
-                    let mut bytes = Vec::<u8>::with_capacity(status.encoded_len());
-                    status.encode(&mut bytes).expect("Failed to encode status.");
-                    bytes
-                };
-                let mut header = MetadataMap::new();
-                header.insert_bin(RW_ERROR_GRPC_HEADER, MetadataValue::from_bytes(&bytes));
-                tonic::Status::with_metadata(Code::Internal, err.to_string(), header)
-            }
+            ErrorCode::InternalError(e) => tonic::Status::internal(e),
+            _ => tonic::Status::internal(err.to_string()),
         }
     }
 }
@@ -386,7 +375,7 @@ impl From<tonic::Status> for RwError {
                 ErrorCode::InvalidParameterValue(err.message().to_string()).into()
             }
             Code::PermissionDenied => ErrorCode::PermissionDenied(err.message().to_string()).into(),
-            _ => ErrorCode::RpcError(err.into()).into(),
+            _ => ErrorCode::InternalError(err.message().to_string()).into(),
         }
     }
 }

src/connector/src/macros.rs

@@ -20,13 +20,25 @@ macro_rules! impl_split_enumerator {
              pub async fn create(properties: ConnectorProperties) -> Result<Self> {
                 match properties {
                     $( ConnectorProperties::$variant_name(props) => $split_enumerator_name::new(props).await.map(Self::$variant_name), )*
-                    other => Err(anyhow!("split enumerator type for config {:?} is not supported", other)),
+                    other => Err(anyhow!(
+                        "split enumerator type for config {:?} is not supported",
+                        other
+                    )),
                 }
              }
 
              pub async fn list_splits(&mut self) -> Result<Vec<SplitImpl>> {
                 match self {
-                    $( Self::$variant_name(inner) => inner.list_splits().await.map(|ss| ss.into_iter().map(SplitImpl::$variant_name).collect_vec()), )*
+                    $( Self::$variant_name(inner) => inner
+                        .list_splits()
+                        .await
+                        .map(|ss| {
+                            ss.into_iter()
+                                .map(SplitImpl::$variant_name)
+                                .collect_vec()
+                        })
+                        .map_err(|e| ErrorCode::ConnectorError(e.to_string()).into()),
+                    )*
                 }
              }
         }

src/connector/src/source/base.rs

@@ -20,6 +20,7 @@ use bytes::Bytes;
 use enum_as_inner::EnumAsInner;
 use itertools::Itertools;
 use prost::Message;
+use risingwave_common::error::ErrorCode;
 use risingwave_pb::source::ConnectorSplit;
 use serde::{Deserialize, Serialize};
 

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

@@ -89,7 +89,13 @@ impl SplitEnumerator for KafkaSplitEnumerator {
     }
 
     async fn list_splits(&mut self) -> anyhow::Result<Vec<KafkaSplit>> {
-        let topic_partitions = self.fetch_topic_partition()?;
+        let topic_partitions = self.fetch_topic_partition().map_err(|e| {
+            anyhow!(
+                "failed to fetch metadata from kafka ({}): {}",
+                self.broker_address,
+                e
+            )
+        })?;
 
         let mut start_offsets = self
             .fetch_start_offset(topic_partitions.as_ref())

src/source/src/parser/common.rs

@@ -23,20 +23,21 @@ use serde_json::Value;
 macro_rules! ensure_float {
     ($v:ident, $t:ty) => {
         $v.as_f64()
-            .ok_or_else(|| anyhow!(concat!("expect ", stringify!($t))))?
+            .ok_or_else(|| anyhow!(concat!("expect ", stringify!($t), ", but found {}"), $v))?
     };
 }
 
 macro_rules! ensure_int {
     ($v:ident, $t:ty) => {
         $v.as_i64()
-            .ok_or_else(|| anyhow!(concat!("expect ", stringify!($t))))?
+            .ok_or_else(|| anyhow!(concat!("expect ", stringify!($t), ", but found {}"), $v))?
     };
 }
 
 macro_rules! ensure_str {
     ($v:ident, $t:literal) => {
-        $v.as_str().ok_or_else(|| anyhow!(concat!("expect ", $t)))?
+        $v.as_str()
+            .ok_or_else(|| anyhow!(concat!("expect ", $t, ", but found {}"), $v))?
     };
 }
 
@@ -81,6 +82,8 @@ fn do_parse_json_value(column: &ColumnDesc, v: &Value) -> Result<ScalarImpl> {
 pub(crate) fn json_parse_value(column: &ColumnDesc, value: Option<&Value>) -> Result<Datum> {
     match value {
         None | Some(Value::Null) => Ok(None),
-        Some(v) => Ok(Some(do_parse_json_value(column, v)?)),
+        Some(v) => Ok(Some(do_parse_json_value(column, v).map_err(|e| {
+            anyhow!("failed to parse column '{}' from json: {}", column.name, e)
+        })?)),
     }
 }