feat(meta): introduce sink validation in meta (#8417)

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkValidationHandler.java

@@ -43,10 +43,13 @@ public void handle(ConnectorServiceProto.ValidateSinkRequest request) {
                     ConnectorServiceProto.ValidateSinkResponse.newBuilder()
                             .setError(
                                     ConnectorServiceProto.ValidationError.newBuilder()
-                                            .setErrorMessage(e.toString())
+                                            .setErrorMessage(e.getMessage())
                                             .build())
                             .build());
             responseObserver.onCompleted();
         }
+
+        responseObserver.onNext(ConnectorServiceProto.ValidateSinkResponse.newBuilder().build());
+        responseObserver.onCompleted();
     }
 }

src/common/src/util/addr.rs

@@ -15,10 +15,9 @@
 use std::net::SocketAddr;
 use std::str::FromStr;
 
+use anyhow::anyhow;
 use risingwave_pb::common::HostAddress as ProstHostAddress;
 
-use crate::error::{internal_error, Result};
-
 /// General host address and port.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct HostAddr {
@@ -41,33 +40,33 @@ impl From<SocketAddr> for HostAddr {
 }
 
 impl TryFrom<&str> for HostAddr {
-    type Error = crate::error::RwError;
+    type Error = anyhow::Error;
 
-    fn try_from(s: &str) -> Result<Self> {
-        let addr = url::Url::parse(&format!("http://{}", s))
-            .map_err(|e| internal_error(format!("{}: {}", e, s)))?;
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        let addr =
+            url::Url::parse(&format!("http://{}", s)).map_err(|e| anyhow!("{}: {}", e, s))?;
         Ok(HostAddr {
             host: addr
                 .host()
-                .ok_or_else(|| internal_error("invalid host"))?
+                .ok_or_else(|| anyhow!("invalid host"))?
                 .to_string(),
-            port: addr.port().ok_or_else(|| internal_error("invalid port"))?,
+            port: addr.port().ok_or_else(|| anyhow!("invalid port"))?,
         })
     }
 }
 
 impl TryFrom<&String> for HostAddr {
-    type Error = crate::error::RwError;
+    type Error = anyhow::Error;
 
-    fn try_from(s: &String) -> Result<Self> {
+    fn try_from(s: &String) -> Result<Self, Self::Error> {
         Self::try_from(s.as_str())
     }
 }
 
 impl FromStr for HostAddr {
-    type Err = crate::error::RwError;
+    type Err = anyhow::Error;
 
-    fn from_str(s: &str) -> Result<Self> {
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
         Self::try_from(s)
     }
 }

src/connector/src/sink/catalog/mod.rs

@@ -17,7 +17,9 @@ pub mod desc;
 use std::collections::HashMap;
 
 use itertools::Itertools;
-use risingwave_common::catalog::{ColumnCatalog, DatabaseId, SchemaId, TableId, UserId};
+use risingwave_common::catalog::{
+    ColumnCatalog, DatabaseId, Field, Schema, SchemaId, TableId, UserId,
+};
 use risingwave_common::util::sort_util::ColumnOrder;
 use risingwave_pb::catalog::{Sink as ProstSink, SinkType as ProstSinkType};
 
@@ -165,6 +167,19 @@ impl SinkCatalog {
             sink_type: self.sink_type.to_proto() as i32,
         }
     }
+
+    pub fn schema(&self) -> Schema {
+        let fields = self
+            .columns
+            .iter()
+            .map(|column| Field::from(column.column_desc.clone()))
+            .collect_vec();
+        Schema { fields }
+    }
+
+    pub fn pk_indices(&self) -> Vec<usize> {
+        self.pk.iter().map(|k| k.column_index).collect_vec()
+    }
 }
 
 impl From<ProstSink> for SinkCatalog {

src/connector/src/sink/mod.rs

@@ -31,7 +31,7 @@ use serde::{Deserialize, Serialize};
 use thiserror::Error;
 pub use tracing;
 
-use self::catalog::SinkType;
+use self::catalog::{SinkCatalog, SinkType};
 use crate::sink::console::{ConsoleConfig, ConsoleSink, CONSOLE_SINK};
 use crate::sink::kafka::{KafkaConfig, KafkaSink, KAFKA_SINK};
 use crate::sink::redis::{RedisConfig, RedisSink};
@@ -160,6 +160,37 @@ impl SinkImpl {
             SinkConfig::BlackHole => SinkImpl::Blackhole,
         })
     }
+
+    pub async fn validate(
+        cfg: SinkConfig,
+        sink_catalog: SinkCatalog,
+        connector_rpc_endpoint: Option<String>,
+    ) -> Result<()> {
+        match cfg {
+            SinkConfig::Redis(cfg) => RedisSink::new(cfg, sink_catalog.schema()).map(|_| ()),
+            SinkConfig::Kafka(cfg) => {
+                // We simply call `KafkaSink::new` here to validate a Kafka sink.
+                if sink_catalog.sink_type.is_append_only() {
+                    KafkaSink::<true>::new(*cfg, sink_catalog.schema(), sink_catalog.pk_indices())
+                        .await
+                        .map(|_| ())
+                } else {
+                    KafkaSink::<false>::new(*cfg, sink_catalog.schema(), sink_catalog.pk_indices())
+                        .await
+                        .map(|_| ())
+                }
+            }
+            SinkConfig::Remote(cfg) => {
+                if sink_catalog.sink_type.is_append_only() {
+                    RemoteSink::<true>::validate(cfg, sink_catalog, connector_rpc_endpoint).await
+                } else {
+                    RemoteSink::<false>::validate(cfg, sink_catalog, connector_rpc_endpoint).await
+                }
+            }
+            SinkConfig::Console(_) => Ok(()),
+            SinkConfig::BlackHole => Ok(()),
+        }
+    }
 }
 
 macro_rules! impl_sink {
@@ -222,7 +253,7 @@ pub enum SinkError {
 
 impl From<RpcError> for SinkError {
     fn from(value: RpcError) -> Self {
-        SinkError::Remote(format!("{:?}", value))
+        SinkError::Remote(format!("{}", value))
     }
 }