refactor(connector): replace validate source rpc with jni

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/JniSourceValidateHandler.java

@@ -0,0 +1,49 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.risingwave.connector.source;
+
+import static com.risingwave.connector.source.SourceValidateHandler.validateResponse;
+import static com.risingwave.connector.source.SourceValidateHandler.validateSource;
+
+import com.risingwave.proto.ConnectorServiceProto;
+import io.grpc.StatusRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JniSourceValidateHandler {
+    static final Logger LOG = LoggerFactory.getLogger(JniSourceValidateHandler.class);
+
+    public static byte[] validate(byte[] validateSourceRequestBytes)
+            throws com.google.protobuf.InvalidProtocolBufferException {
+        try {
+            var request =
+                    ConnectorServiceProto.ValidateSourceRequest.parseFrom(
+                            validateSourceRequestBytes);
+
+            // For jni.rs
+            java.lang.Thread.currentThread()
+                    .setContextClassLoader(java.lang.ClassLoader.getSystemClassLoader());
+            validateSource(request);
+            // validate pass
+            return ConnectorServiceProto.ValidateSourceResponse.newBuilder().build().toByteArray();
+        } catch (StatusRuntimeException e) {
+            LOG.warn("Source validation failed", e);
+            return validateResponse(e.getMessage()).toByteArray();
+        } catch (Exception e) {
+            LOG.error("Internal error on source validation", e);
+            return validateResponse("Internal error: " + e.getMessage()).toByteArray();
+        }
+    }
+}

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/SourceValidateHandler.java

@@ -56,7 +56,7 @@ public void handle(ConnectorServiceProto.ValidateSourceRequest request) {
         }
     }
 
-    private ConnectorServiceProto.ValidateSourceResponse validateResponse(String message) {
+    public static ConnectorServiceProto.ValidateSourceResponse validateResponse(String message) {
         return ConnectorServiceProto.ValidateSourceResponse.newBuilder()
                 .setError(
                         ConnectorServiceProto.ValidationError.newBuilder()
@@ -65,14 +65,14 @@ private ConnectorServiceProto.ValidateSourceResponse validateResponse(String mes
                 .build();
     }
 
-    private void ensurePropNotNull(Map<String, String> props, String name) {
+    public static void ensurePropNotNull(Map<String, String> props, String name) {
         if (!props.containsKey(name)) {
             throw ValidatorUtils.invalidArgument(
                     String.format("'%s' not found, please check the WITH properties", name));
         }
     }
 
-    private void validateSource(ConnectorServiceProto.ValidateSourceRequest request)
+    public static void validateSource(ConnectorServiceProto.ValidateSourceRequest request)
             throws Exception {
         var props = request.getPropertiesMap();
 

src/connector/src/source/cdc/enumerator/mod.rs

@@ -13,13 +13,17 @@
 // limitations under the License.
 
 use std::marker::PhantomData;
+use std::ops::Deref;
 use std::str::FromStr;
 
 use anyhow::anyhow;
 use async_trait::async_trait;
 use itertools::Itertools;
+use jni::objects::{JByteArray, JValue, JValueOwned};
+use prost::Message;
 use risingwave_common::util::addr::HostAddr;
-use risingwave_pb::connector_service::SourceType;
+use risingwave_jni_core::jvm_runtime::JVM;
+use risingwave_pb::connector_service::{SourceType, ValidateSourceRequest, ValidateSourceResponse};
 
 use crate::source::cdc::{
     CdcProperties, CdcSourceTypeTrait, CdcSplitBase, Citus, DebeziumCdcSplit, MySqlCdcSplit, Mysql,
@@ -49,10 +53,6 @@ where
         props: CdcProperties<T>,
         context: SourceEnumeratorContextRef,
     ) -> anyhow::Result<Self> {
-        let connector_client = context.connector_client.clone().ok_or_else(|| {
-            anyhow!("connector node endpoint not specified or unable to connect to connector node")
-        })?;
-
         let server_addrs = props
             .props
             .get(DATABASE_SERVERS_KEY)
@@ -69,15 +69,43 @@ where
             SourceType::from(T::source_type())
         );
 
+        let mut env = JVM.as_ref()?.attach_current_thread()?;
+
+        let validate_source_request = ValidateSourceRequest {
+            source_id: context.info.source_id as u64,
+            source_type: props.get_source_type_pb() as _,
+            properties: props.props,
+            table_schema: Some(props.table_schema),
+        };
+
+        let validate_source_request_bytes = env
+            .byte_array_from_slice(&Message::encode_to_vec(&validate_source_request))
+            .unwrap();
+
         // validate connector properties
-        connector_client
-            .validate_source_properties(
-                context.info.source_id as u64,
-                props.get_source_type_pb(),
-                props.props,
-                Some(props.table_schema),
-            )
-            .await?;
+        let response = env.call_static_method(
+            "com/risingwave/connector/source/JniSourceValidateHandler",
+            "validate",
+            "([B)[B",
+            &[JValue::Object(&validate_source_request_bytes)],
+        )?;
+
+        let validate_source_response_bytes = match response {
+            JValueOwned::Object(o) => unsafe { JByteArray::from_raw(o.into_raw()) },
+            _ => unreachable!(),
+        };
+
+        let validate_source_response: ValidateSourceResponse = Message::decode(
+            risingwave_jni_core::to_guarded_slice(&validate_source_response_bytes, &mut env)?
+                .deref(),
+        )?;
+
+        validate_source_response.error.map_or(Ok(()), |err| {
+            Err(anyhow!(format!(
+                "source cannot pass validation: {}",
+                err.error_message
+            )))
+        })?;
 
         tracing::debug!("validate cdc source properties success");
         Ok(Self {

src/jni_core/src/lib.rs

@@ -57,7 +57,7 @@ pub type GetEventStreamJniSender = Sender<GetEventStreamResponse>;
 static RUNTIME: LazyLock<Runtime> = LazyLock::new(|| tokio::runtime::Runtime::new().unwrap());
 
 #[derive(Error, Debug)]
-enum BindingError {
+pub enum BindingError {
     #[error("JniError {error}")]
     Jni {
         #[from]
@@ -89,7 +89,7 @@ enum BindingError {
 
 type Result<T> = std::result::Result<T, BindingError>;
 
-fn to_guarded_slice<'array, 'env>(
+pub fn to_guarded_slice<'array, 'env>(
     array: &'array JByteArray<'env>,
     env: &'array mut JNIEnv<'env>,
 ) -> Result<SliceGuard<'env, 'array>> {