feat(meta): introduce sink validation in meta

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

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

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

@@ -20,7 +20,7 @@ use itertools::Itertools;
 use risingwave_common::catalog::{
     ColumnCatalog, DatabaseId, Field, Schema, SchemaId, TableId, UserId,
 };
-use risingwave_common::util::sort_util::OrderPair;
+use risingwave_common::util::sort_util::ColumnOrder;
 use risingwave_pb::catalog::{Sink as ProstSink, SinkType as ProstSinkType};
 
 #[derive(Clone, Copy, Debug, Default, Hash, PartialOrd, PartialEq, Eq)]
@@ -178,7 +178,7 @@ impl SinkCatalog {
     }
 
     pub fn pk_indices(&self) -> Vec<usize> {
-        self.pk.iter().map(|k| k.column_idx).collect_vec()
+        self.pk.iter().map(|k| k.column_index).collect_vec()
     }
 }
 

src/connector/src/sink/kafka.rs

@@ -63,7 +63,6 @@ pub struct KafkaConfig {
 
     pub format: String, // accept "append_only", "debezium", or "upsert"
 
-    #[serde(default)]
     pub identifier: String,
 
     #[serde(

src/connector/src/sink/mod.rs

@@ -253,7 +253,7 @@ pub enum SinkError {
 
 impl From<RpcError> for SinkError {
     fn from(value: RpcError) -> Self {
-        SinkError::Remote(format!("{:?}", value))
+        SinkError::Remote(format!("{}", value))
     }
 }
 

src/meta/src/manager/streaming_job.rs

@@ -151,26 +151,4 @@ impl StreamingJob {
             None
         }
     }
-
-    /// Returns the optional [`Source`] if this is a `Table` streaming job.
-    ///
-    /// Only used for registering sources for creating tables with connectors.
-    pub fn source(&self) -> Option<&Source> {
-        match self {
-            Self::MaterializedView(_) => None,
-            Self::Sink(_) => None,
-            Self::Table(source, _) => source.as_ref(),
-            Self::Index(_, _) => None,
-        }
-    }
-
-    /// Returns the [`Sink`] if this is a `Sink` streaming job.
-    pub fn sink(&self) -> Option<&Sink> {
-        match self {
-            Self::MaterializedView(_) => None,
-            Self::Sink(sink) => Some(sink),
-            Self::Table(_, _) => None,
-            Self::Index(_, _) => None,
-        }
-    }
 }

src/meta/src/rpc/ddl_controller.rs

@@ -237,13 +237,18 @@ where
 
             internal_tables = ctx.internal_tables();
 
-            if let Some(source) = stream_job.source() {
-                // Register the source on the connector node.
-                self.source_manager.register_source(source).await?;
-            } else if let Some(sink) = stream_job.sink() {
-                // Validate the sink on the connector node.
-                validate_sink(sink, self.env.opts.connector_rpc_endpoint.clone()).await?;
+            match &stream_job {
+                StreamingJob::Table(Some(source), _) => {
+                    // Register the source on the connector node.
+                    self.source_manager.register_source(source).await?;
+                }
+                StreamingJob::Sink(sink) => {
+                    // Validate the sink on the connector node.
+                    validate_sink(sink, self.env.opts.connector_rpc_endpoint.clone()).await?;
+                }
+                _ => {}
             }
+
             self.stream_manager
                 .create_streaming_job(table_fragments, ctx)
                 .await?;

src/meta/src/stream/sink.rs

@@ -24,8 +24,12 @@ pub async fn validate_sink(
     connector_rpc_endpoint: Option<String>,
 ) -> MetaResult<()> {
     let sink_catalog = SinkCatalog::from(prost_sink_catalog);
-    let sink_config = SinkConfig::from_hashmap(sink_catalog.properties.clone())
+    let mut properties = sink_catalog.properties.clone();
+    // Insert a value as the `identifier` field to get parsed by serde.
+    properties.insert("identifier".to_string(), u64::MAX.to_string());
+    let sink_config = SinkConfig::from_hashmap(properties)
         .map_err(|err| MetaError::from(anyhow!(err.to_string())))?;
+
     SinkImpl::validate(sink_config, sink_catalog, connector_rpc_endpoint)
         .await
         .map_err(|err| MetaError::from(anyhow!(err.to_string())))

src/rpc_client/src/connector_client.rs

@@ -169,7 +169,12 @@ impl ConnectorClient {
             .into_inner();
         response.error.map_or_else(
             || Ok(()), // If there is no error message, return Ok here.
-            |err| Err(RpcError::Internal(anyhow!(err.error_message))),
+            |err| {
+                Err(RpcError::Internal(anyhow!(format!(
+                    "sink cannot pass validation: {}",
+                    err.error_message
+                ))))
+            },
         )
     }
 }