risingwave_connector/
error.rs1use std::sync::Arc;
16
17use risingwave_common::array::ArrayError;
18use risingwave_common::error::def_anyhow_newtype;
19use risingwave_pb::PbFieldNotFound;
20use risingwave_rpc_client::error::RpcError;
21
22use crate::parser::AccessError;
23use crate::schema::InvalidOptionError;
24use crate::schema::schema_registry::{ConcurrentRequestError, WireFormatError};
25use crate::sink::SinkError;
26use crate::source::mqtt::MqttError;
27use crate::source::nats::NatsJetStreamError;
28
29def_anyhow_newtype! {
30 pub ConnectorError,
31
32 std::io::Error => transparent,
34 Arc<ConnectorError> => transparent,
35
36 AccessError => transparent,
38 WireFormatError => transparent,
39 ConcurrentRequestError => transparent,
40 InvalidOptionError => transparent,
41 SinkError => transparent,
42 PbFieldNotFound => transparent,
43
44 url::ParseError => "failed to parse url",
48 serde_json::Error => "failed to parse json",
49 csv::Error => "failed to parse csv",
50
51 uuid::Error => transparent, opendal::Error => transparent, parquet::errors::ParquetError => transparent,
56 ArrayError => "Array error",
57 sqlx::Error => transparent, mysql_async::Error => "MySQL error",
59 tokio_postgres::Error => "Postgres error",
60 tiberius::error::Error => "Sql Server error",
61 apache_avro::Error => "Avro error",
62 rdkafka::error::KafkaError => "Kafka error",
63 pulsar::Error => "Pulsar error",
64
65 async_nats::jetstream::consumer::StreamError => "Nats error",
66 async_nats::jetstream::consumer::pull::MessagesError => "Nats error",
67 async_nats::jetstream::context::CreateStreamError => "Nats error",
68 async_nats::jetstream::stream::ConsumerError => "Nats error",
69 async_nats::error::Error<async_nats::jetstream::context::RequestErrorKind> => "Nats error",
70 NatsJetStreamError => "Nats error",
71
72 iceberg::Error => "IcebergV2 error",
73 redis::RedisError => "Redis error",
74 risingwave_common::array::arrow::arrow_schema_iceberg::ArrowError => "Arrow error",
75 google_cloud_pubsub::client::google_cloud_auth::error::Error => "Google Cloud error",
76 rumqttc::tokio_rustls::rustls::Error => "TLS error",
77 rumqttc::v5::ClientError => "MQTT SDK error",
78 rumqttc::v5::OptionError => "MQTT Option error",
79 MqttError => "MQTT Source error",
80 mongodb::error::Error => "Mongodb error",
81
82 openssl::error::ErrorStack => "OpenSSL error",
83 risingwave_common::secret::SecretError => "Secret error",
84}
85
86pub type ConnectorResult<T, E = ConnectorError> = std::result::Result<T, E>;
87
88impl From<ConnectorError> for RpcError {
89 fn from(value: ConnectorError) -> Self {
90 RpcError::Internal(value.0)
91 }
92}