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

docker/Dockerfile

@@ -64,6 +64,8 @@ COPY --from=builder /risingwave/bin/jeprof /usr/local/bin/jeprof
 ENV PLAYGROUND_PROFILE docker-playground
 # Set default dashboard UI to local path instead of github proxy
 ENV RW_DASHBOARD_UI_PATH /risingwave/ui
+# Set default connector libs path
+ENV CONNECTOR_LIBS_PATH /risingwave/bin/connector-node/libs
 
 ENTRYPOINT [ "/risingwave/bin/risingwave" ]
 CMD [ "playground" ]

docker/Dockerfile.hdfs

@@ -88,6 +88,8 @@ ENV CLASSPATH ${HADOOP_CONF_DIR}:${CLASSPATH}
 ENV PLAYGROUND_PROFILE docker-playground
 # Set default dashboard UI to local path instead of github proxy
 ENV RW_DASHBOARD_UI_PATH /risingwave/ui
+# Set default connector libs path
+ENV CONNECTOR_LIBS_PATH /risingwave/bin/connector-node/libs
 
 ENTRYPOINT [ "/risingwave/hdfs_env.sh" ]
 CMD [ "playground" ]

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,107 @@
+// 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.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 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);
+    }
+
+    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());
+
+            while (runner.isRunning()) {
+                // 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 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"] }
@@ -83,6 +84,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"