feat(connector): init embedded connector node

ci/scripts/e2e-source-test.sh

@@ -7,6 +7,7 @@ source ci/scripts/common.sh
 
 # prepare environment
 export CONNECTOR_RPC_ENDPOINT="localhost:50051"
+export CONNECTOR_LIBS_PATH="./connector-node/libs"
 
 while getopts 'p:' opt; do
     case ${opt} in

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

@@ -70,6 +70,33 @@ public static CdcEngineRunner newCdcEngineRunner(
         return runner;
     }
 
+    public static CdcEngineRunner newCdcEngineRunner(DbzConnectorConfig config) {
+        DbzCdcEngineRunner runner = null;
+        try {
+            var sourceId = config.getSourceId();
+            var engine =
+                    new DbzCdcEngine(
+                            config.getSourceId(),
+                            config.getResolvedDebeziumProps(),
+                            (success, message, error) -> {
+                                if (!success) {
+                                    LOG.error(
+                                            "engine#{} terminated with error. message: {}",
+                                            sourceId,
+                                            message,
+                                            error);
+                                } else {
+                                    LOG.info("engine#{} stopped normally. {}", sourceId, message);
+                                }
+                            });
+
+            runner = new DbzCdcEngineRunner(engine);
+        } catch (Exception e) {
+            LOG.error("failed to create the CDC engine", e);
+        }
+        return runner;
+    }
+
     /** Start to run the cdc engine */
     public void start() {
         if (isRunning()) {

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

@@ -0,0 +1,132 @@
+// 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.core;
+
+import com.risingwave.connector.api.source.CdcEngineRunner;
+import com.risingwave.connector.api.source.SourceTypeE;
+import com.risingwave.connector.source.common.DbzConnectorConfig;
+import com.risingwave.java.binding.Binding;
+import com.risingwave.metrics.ConnectorNodeMetrics;
+import com.risingwave.proto.ConnectorServiceProto;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** handler for starting a debezium source connectors for jni */
+public class JniDbzSourceHandler {
+    static final Logger LOG = LoggerFactory.getLogger(DbzSourceHandler.class);
+
+    private final DbzConnectorConfig config;
+
+    public JniDbzSourceHandler(DbzConnectorConfig config) {
+        this.config = config;
+    }
+
+    public static void runJniDbzSourceThread(byte[] getEventStreamRequestBytes, long channelPtr)
+            throws com.google.protobuf.InvalidProtocolBufferException {
+
+        var request =
+                ConnectorServiceProto.GetEventStreamRequest.parseFrom(getEventStreamRequestBytes);
+
+        // For jni.rs
+        java.lang.Thread.currentThread()
+                .setContextClassLoader(java.lang.ClassLoader.getSystemClassLoader());
+        // userProps extracted from grpc request, underlying implementation is UnmodifiableMap
+        Map<String, String> mutableUserProps = new HashMap<>(request.getPropertiesMap());
+        mutableUserProps.put("source.id", Long.toString(request.getSourceId()));
+        var config =
+                new DbzConnectorConfig(
+                        SourceTypeE.valueOf(request.getSourceType()),
+                        request.getSourceId(),
+                        request.getStartOffset(),
+                        mutableUserProps,
+                        request.getSnapshotDone());
+        JniDbzSourceHandler handler = new JniDbzSourceHandler(config);
+        handler.start(channelPtr);
+    }
+
+    class OnReadyHandler implements Runnable {
+        private final CdcEngineRunner runner;
+        private final long channelPtr;
+
+        public OnReadyHandler(CdcEngineRunner runner, long channelPtr) {
+            this.runner = runner;
+            this.channelPtr = channelPtr;
+        }
+
+        @Override
+        public void run() {
+            while (runner.isRunning()) {
+                try {
+                    // check whether the send queue has room for new messages
+                    // Thread will block on the channel to get output from engine
+                    var resp =
+                            runner.getEngine().getOutputChannel().poll(500, TimeUnit.MILLISECONDS);
+                    boolean success;
+                    if (resp != null) {
+                        ConnectorNodeMetrics.incSourceRowsReceived(
+                                config.getSourceType().toString(),
+                                String.valueOf(config.getSourceId()),
+                                resp.getEventsCount());
+                        LOG.info(
+                                "Engine#{}: emit one chunk {} events to network ",
+                                config.getSourceId(),
+                                resp.getEventsCount());
+                        success = Binding.sendCdcSourceMsgToChannel(channelPtr, resp.toByteArray());
+                    } else {
+                        // If resp is null means just check whether channel is closed.
+                        success = Binding.sendCdcSourceMsgToChannel(channelPtr, null);
+                    }
+                    if (!success) {
+                        LOG.info(
+                                "Engine#{}: JNI sender broken detected, stop the engine",
+                                config.getSourceId());
+                        runner.stop();
+                        return;
+                    }
+                } catch (Throwable e) {
+                    LOG.error("Poll engine output channel fail. ", e);
+                }
+            }
+        }
+    }
+
+    public void start(long channelPtr) {
+        var runner = DbzCdcEngineRunner.newCdcEngineRunner(config);
+        if (runner == null) {
+            return;
+        }
+
+        try {
+            // Start the engine
+            runner.start();
+            LOG.info("Start consuming events of table {}", config.getSourceId());
+
+            final OnReadyHandler onReadyHandler = new OnReadyHandler(runner, channelPtr);
+
+            onReadyHandler.run();
+
+        } catch (Throwable t) {
+            LOG.error("Cdc engine failed.", t);
+            try {
+                runner.stop();
+            } catch (Exception e) {
+                LOG.warn("Failed to stop Engine#{}", config.getSourceId(), e);
+            }
+        }
+    }
+}

java/java-binding/src/main/java/com/risingwave/java/binding/Binding.java

@@ -17,8 +17,13 @@
 import io.questdb.jar.jni.JarJniLoader;
 
 public class Binding {
+    private static final boolean IS_EMBEDDED_CONNECTOR =
+            Boolean.parseBoolean(System.getProperty("is_embedded_connector"));
+
     static {
-        JarJniLoader.loadLib(Binding.class, "/risingwave/jni", "risingwave_java_binding");
+        if (!IS_EMBEDDED_CONNECTOR) {
+            JarJniLoader.loadLib(Binding.class, "/risingwave/jni", "risingwave_java_binding");
+        }
     }
 
     public static native int vnodeCount();
@@ -84,4 +89,6 @@ public class Binding {
     static native void streamChunkIteratorClose(long pointer);
 
     static native long streamChunkIteratorFromPretty(String str);
+
+    public static native boolean sendCdcSourceMsgToChannel(long channelPtr, byte[] msg);
 }

src/connector/Cargo.toml

@@ -53,6 +53,7 @@ hyper = "0.14"
 hyper-tls = "0.5"
 icelake = { workspace = true }
 itertools = "0.11"
+jni = { version = "0.21.1", features = ["invocation"] }
 jsonschema-transpiler = "1.10.0"
 maplit = "1.0.2"
 moka = { version = "0.11", features = ["future"] }
@@ -82,6 +83,7 @@ rdkafka = { workspace = true, features = [
 ] }
 reqwest = { version = "0.11", features = ["json"] }
 risingwave_common = { workspace = true }
+risingwave_jni_core = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 rust_decimal = "1"

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

@@ -18,6 +18,7 @@ use anyhow::anyhow;
 use async_trait::async_trait;
 use itertools::Itertools;
 use risingwave_common::util::addr::HostAddr;
+use risingwave_jni_core::jvm_runtime::JVM;
 use risingwave_pb::connector_service::SourceType as PbSourceType;
 
 use crate::source::cdc::{
@@ -60,6 +61,7 @@ impl SplitEnumerator for DebeziumSplitEnumerator {
             .unwrap_or_default();
 
         let source_type = props.get_source_type_pb()?;
+        JVM.as_ref()?;
         // validate connector properties
         connector_client
             .validate_source_properties(