feat(connector): init embedded connector node

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/JniDbzSourceHandler.java

@@ -17,7 +17,6 @@
 import com.risingwave.connector.api.source.SourceTypeE;
 import com.risingwave.connector.source.common.DbzConnectorConfig;
 import com.risingwave.java.binding.Binding;
-import com.risingwave.java.binding.CdcJniChannel;
 import com.risingwave.metrics.ConnectorNodeMetrics;
 import com.risingwave.proto.ConnectorServiceProto;
 import java.util.HashMap;
@@ -38,27 +37,24 @@ public JniDbzSourceHandler(DbzConnectorConfig config) {
 
     public static void runJniDbzSourceThread(byte[] getEventStreamRequestBytes, long channelPtr)
             throws com.google.protobuf.InvalidProtocolBufferException {
-        try (CdcJniChannel channel = new CdcJniChannel(channelPtr)) {
-            var request =
-                    ConnectorServiceProto.GetEventStreamRequest.parseFrom(
-                            getEventStreamRequestBytes);
+        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(channel.getPointer());
-        }
+        // 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) {

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

@@ -91,6 +91,4 @@ public class Binding {
     static native long streamChunkIteratorFromPretty(String str);
 
     public static native boolean sendCdcSourceMsgToChannel(long channelPtr, byte[] msg);
-
-    static native void cdcJniChannelClose(long pointer);
 }

src/connector/src/source/cdc/source/reader.rs

@@ -151,24 +151,33 @@ where
 
             let channel_ptr = Box::into_raw(tx) as i64;
 
-            let _ = env
-                .call_static_method(
-                    "com/risingwave/connector/source/core/JniDbzSourceHandler",
-                    "runJniDbzSourceThread",
-                    "([BJ)V",
-                    &[
-                        JValue::Object(&get_event_stream_request_bytes),
-                        JValue::from(channel_ptr),
-                    ],
-                )
-                .inspect_err(|e| tracing::error!("jni call error: {:?}", e))
-                .unwrap();
+            let result = env.call_static_method(
+                "com/risingwave/connector/source/core/JniDbzSourceHandler",
+                "runJniDbzSourceThread",
+                "([BJ)V",
+                &[
+                    JValue::Object(&get_event_stream_request_bytes),
+                    JValue::from(channel_ptr),
+                ],
+            );
+
+            unsafe { drop(Box::from_raw(channel_ptr as *mut GetEventStreamJniSender)) };
+            match result {
+                Ok(_) => {
+                    tracing::info!("end of jni call runJniDbzSourceThread");
+                }
+                Err(e) => {
+                    tracing::error!("jni call error: {:?}", e);
+                }
+            }
         });
 
         while let Some(GetEventStreamResponse { events, .. }) = rx.recv().await {
             tracing::debug!("receive events {:?}", events.len());
             let msgs = events.into_iter().map(SourceMessage::from).collect_vec();
             yield msgs;
         }
+
+        Err(anyhow!("all senders are dropped"))?;
     }
 }

src/jni_core/src/jvm_runtime.rs

@@ -275,12 +275,6 @@ fn register_native_method_for_jvm(jvm: &JavaVM) {
                 fn_ptr: crate::Java_com_risingwave_java_binding_Binding_sendCdcSourceMsgToChannel
                     as *mut c_void,
             },
-            NativeMethod {
-                name: JNIString::from("cdcJniChannelClose"),
-                sig: JNIString::from("(J)V"),
-                fn_ptr: crate::Java_com_risingwave_java_binding_Binding_cdcJniChannelClose
-                    as *mut c_void,
-            },
         ],
     )
     .inspect_err(|e| tracing::error!("jvm register native methods error: {:?}", e))

src/jni_core/src/lib.rs

@@ -864,14 +864,6 @@ pub extern "system" fn Java_com_risingwave_java_binding_Binding_sendCdcSourceMsg
     })
 }
 
-#[no_mangle]
-pub extern "system" fn Java_com_risingwave_java_binding_Binding_cdcJniChannelClose<'a>(
-    _env: EnvParam<'a>,
-    pointer: Pointer<'a, GetEventStreamJniSender>,
-) {
-    pointer.drop()
-}
-
 #[cfg(test)]
 mod tests {
     use risingwave_common::types::{DataType, Timestamptz};