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