use SinkError::Config

risingwavelabs · Sep 29, 2023 · c1291f1 · c1291f1
1 parent 0610481
commit c1291f1
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/src/connector/src/sink/catalog/mod.rs b/src/connector/src/sink/catalog/mod.rs
@@ -180,7 +180,7 @@ impl SinkFormatDesc {
 }
 
 impl TryFrom<PbSinkFormatDesc> for SinkFormatDesc {
-    type Error = String;
+    type Error = SinkError;
 
     fn try_from(value: PbSinkFormatDesc) -> Result<Self, Self::Error> {
         use risingwave_pb::plan_common::{EncodeType as E, FormatType as F};
@@ -190,15 +190,21 @@ impl TryFrom<PbSinkFormatDesc> for SinkFormatDesc {
             F::Upsert => SinkFormat::Upsert,
             F::Debezium => SinkFormat::Debezium,
             f @ (F::Unspecified | F::Native | F::DebeziumMongo | F::Maxwell | F::Canal) => {
-                return Err(format!("sink format unsupported: {}", f.as_str_name()))
+                return Err(SinkError::Config(anyhow!(
+                    "sink format unsupported: {}",
+                    f.as_str_name()
+                )))
             }
         };
         let encode = match value.encode() {
             E::Json => SinkEncode::Json,
             E::Protobuf => SinkEncode::Protobuf,
             E::Avro => SinkEncode::Avro,
             e @ (E::Unspecified | E::Native | E::Csv | E::Bytes) => {
-                return Err(format!("sink encode unsupported: {}", e.as_str_name()))
+                return Err(SinkError::Config(anyhow!(
+                    "sink encode unsupported: {}",
+                    e.as_str_name()
+                )))
             }
         };
         let options = value.options.into_iter().collect();

diff --git a/src/stream/src/from_proto/sink.rs b/src/stream/src/from_proto/sink.rs
@@ -82,7 +82,7 @@ impl ExecutorBuilder for SinkExecutorBuilder {
         }?;
         let format_desc = match &sink_desc.format_desc {
             // Case A: new syntax `format ... encode ...`
-            Some(f) => Some(f.clone().try_into().map_err(|e| anyhow!("{e}"))?),
+            Some(f) => Some(f.clone().try_into()?),
             None => match sink_desc.properties.get(SINK_TYPE_OPTION) {
                 // Case B: old syntax `type = '...'`
                 Some(t) => SinkFormatDesc::from_legacy_type(connector, t)?,