test(mongodb-cdc): add integration and e2e test

ci/docker-compose.yml

@@ -77,6 +77,9 @@ services:
       - db
       - message_queue
       - schemaregistry
+      - mongodb
+      - mongodb-setup
+      - mongo_data_generator
     volumes:
       - ..:/risingwave
 
@@ -274,3 +277,42 @@ services:
       interval: 5s
       timeout: 5s
       retries: 5
+
+  mongodb:
+    image: mongo:4.4
+    ports:
+      - "27017"
+    command: --replSet rs0 --oplogSize 128
+    restart: always
+    healthcheck:
+      test: "echo 'db.runCommand({ping: 1})' | mongo"
+      interval: 5s
+      timeout: 10s
+      retries: 3
+
+  mongodb-setup:
+    image: mongo:4.4
+    container_name: mongodb-setup
+    depends_on:
+      - mongodb
+    entrypoint:
+      [
+        "bash",
+        "-c",
+        "sleep 10 && mongo --host mongodb:27017 /config-replica.js && sleep 10"
+      ]
+    restart: "no"
+    volumes:
+      - ./mongodb/config-replica.js:/config-replica.js
+
+  mongo_data_generator:
+      build:
+        context: .
+        dockerfile: ./mongodb/Dockerfile.generator
+      container_name: mongo_data_generator
+      depends_on:
+        - mongodb
+      environment:
+        MONGO_HOST: mongodb
+        MONGO_PORT: 27017
+        MONGO_DB_NAME: random_data

ci/mongodb/Dockerfile.generator

@@ -0,0 +1,23 @@
+# Use an official Python runtime as a parent image
+FROM python:3.8-slim
+
+# Set the working directory to /app
+WORKDIR /app
+
+# Copy the requirements file into the container at /app
+COPY ./mongodb/requirements.txt /app
+COPY ./mongodb/app.py /app
+
+# Install any needed packages specified in requirements.txt
+RUN pip install -r requirements.txt
+
+# Make port 5000 available to the world outside this container
+EXPOSE 5000
+
+# Define environment variables
+ENV MONGO_HOST mongodb
+ENV MONGO_PORT 27017
+ENV MONGO_DB_NAME random_data
+
+# Run app.py when the container launches
+CMD ["python", "app.py"]

ci/mongodb/app.py

@@ -0,0 +1,41 @@
+import os
+import pymongo
+from faker import Faker
+
+# To check the data through mongosh or mongo, run the following command:
+#  > mongosh mongodb://admin:[email protected]:27017
+#  > rs0 [direct: primary] test> use random_data
+#  > rs0 [direct: primary] random_data> db.users.find()
+#  > rs0 [direct: primary] random_data> db.users.count()
+
+# Connect to MongoDB
+mongo_host = os.environ["MONGO_HOST"]
+mongo_port = os.environ["MONGO_PORT"]
+mongo_db_name = os.environ["MONGO_DB_NAME"]
+
+url = f"mongodb://{mongo_host}:{mongo_port}"
+client = pymongo.MongoClient(url)
+db = client[mongo_db_name]
+
+# Generate random data
+fake = Faker()
+collection = db["users"]
+
+for _ in range(55):
+    user_data = {
+        "name": fake.name(),
+        "address": fake.address(),
+        "email": fake.email(),
+    }
+    collection.insert_one(user_data)
+
+# Count the number of records in the collection
+total_records = collection.count_documents({})
+
+# Close the MongoDB connection
+client.close()
+print(f"Random data generated and inserted into MongoDB: {url}")
+
+# Print insertion summary
+print("Insertion summary:")
+print(f"Total records in the collection: {total_records}")

ci/scripts/e2e-source-test.sh

@@ -61,6 +61,20 @@ echo "--- starting risingwave cluster"
 RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
 cargo make ci-start ci-1cn-1fe-with-recovery
 
+echo "--- mongodb cdc test"
+# install the mongo shell
+wget http://archive.ubuntu.com/ubuntu/pool/main/o/openssl/libssl1.1_1.1.1f-1ubuntu2_amd64.deb
+wget https://repo.mongodb.org/apt/ubuntu/dists/focal/mongodb-org/4.4/multiverse/binary-amd64/mongodb-org-shell_4.4.28_amd64.deb
+dpkg -i libssl1.1_1.1.1f-1ubuntu2_amd64.deb
+dpkg -i mongodb-org-shell_4.4.28_amd64.deb
+
+echo '> ping mongodb'
+echo 'db.runCommand({ping: 1})' | mongo mongodb://mongodb:27017
+echo '> rs config'
+echo 'rs.conf()' | mongo mongodb://mongodb:27017
+echo '> run test..'
+sqllogictest -p 4566 -d dev './e2e_test/source/cdc/mongodb/**/*.slt'
+
 echo "--- inline cdc test"
 export MYSQL_HOST=mysql MYSQL_TCP_PORT=3306 MYSQL_PWD=123456
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc_inline/**/*.slt'

ci/scripts/gen-integration-test-yaml.py

@@ -12,6 +12,7 @@
     'schema-registry': ['json'],
     'mysql-cdc': ['json'],
     'postgres-cdc': ['json'],
+    'mongodb-cdc': ['json'],
     'mysql-sink': ['json'],
     'postgres-sink': ['json'],
     'iceberg-cdc': ['json'],

e2e_test/source/cdc/mongodb/mongodb_basic.slt

@@ -0,0 +1,28 @@
+# CDC source basic test
+control substitution on
+
+statement ok
+CREATE TABLE users (_id JSONB PRIMARY KEY, payload JSONB) WITH (
+  connector = 'mongodb-cdc',
+  mongodb.url = 'mongodb://mongodb:27017/?replicaSet=rs0',
+  collection.name = 'random_data.*'
+);
+
+statement ok
+CREATE MATERIALIZED VIEW normalized_users AS
+SELECT
+    payload ->> 'name' as name,
+    payload ->> 'email' as email,
+    payload ->> 'address' as address
+FROM
+    users;
+
+sleep 5s
+
+query I
+select count(*) from normalized_users;
+----
+55
+
+statement ok
+DROP TABLE users cascade

integration_tests/debezium-mongo/docker-compose.yaml

@@ -28,32 +28,12 @@ services:
       service: message_queue
 
   mongodb:
-    image: mongo:4.4
-    container_name: mongodb
-    ports:
-      - "27017:27017"
-    command: --replSet rs0 --oplogSize 128
-    restart: always
-    healthcheck:
-      test: "echo 'db.runCommand({ping: 1})' | mongo"
-      interval: 5s
-      timeout: 10s
-      retries: 3
+    extends: ../mongodb/docker-compose.yaml
+    service: mongodb
 
   mongodb-setup:
-    image: mongo:4.4
-    container_name: mongodb-setup
-    depends_on:
-      - mongodb
-    entrypoint:
-      [
-        "bash",
-        "-c",
-        "sleep 10 && mongo --host mongodb:27017 /config-replica.js && sleep 10"
-      ]
-    restart: "no"
-    volumes:
-      - ./config-replica.js:/config-replica.js
+    extends: ../mongodb/docker-compose.yaml
+    service: mongodb-setup
 
   debezium:
     image: debezium/connect:1.9
@@ -78,16 +58,9 @@ services:
       CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://message_queue:8081
 
   random_data_generator:
-    build:
-      context: .
-      dockerfile: Dockerfile.generator
-    container_name: random_data_generator
-    depends_on:
-      - mongodb
-    environment:
-      MONGO_HOST: mongodb
-      MONGO_PORT: 27017
-      MONGO_DB_NAME: random_data
+    extends: ../mongodb/docker-compose.yaml
+    service: random_data_generator
+
 
   register-mongodb-connector:
     image: curlimages/curl:7.79.1

integration_tests/mongodb-cdc/create_mv.sql

@@ -0,0 +1,7 @@
+CREATE MATERIALIZED VIEW normalized_users AS
+SELECT
+    payload ->> 'name' as name,
+    payload ->> 'email' as email,
+    payload ->> 'address' as address
+FROM
+    users;

integration_tests/mongodb-cdc/create_source.sql

@@ -0,0 +1,5 @@
+CREATE TABLE users (_id JSONB PRIMARY KEY, payload JSONB) WITH (
+    connector = 'mongodb-cdc',
+    mongodb.url = 'mongodb://mongodb:27017/?replicaSet=rs0',
+    collection.name = 'random_data.*'
+);

integration_tests/mongodb-cdc/data_check

@@ -0,0 +1 @@
+users,normalized_users

integration_tests/mongodb-cdc/docker-compose.yaml

@@ -0,0 +1,48 @@
+---
+version: "3"
+services:
+  risingwave-standalone:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: risingwave-standalone
+  etcd-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: etcd-0
+  grafana-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: grafana-0
+  minio-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: minio-0
+  prometheus-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: prometheus-0
+  mongodb:
+    extends:
+      file: ../mongodb/docker-compose.yaml
+      service: mongodb
+  mongodb-setup:
+    extends:
+      file: ../mongodb/docker-compose.yaml
+      service: mongodb-setup
+  random_data_generator:
+    extends:
+      file: ../mongodb/docker-compose.yaml
+      service: random_data_generator
+
+volumes:
+  risingwave-standalone:
+    external: false
+  etcd-0:
+    external: false
+  grafana-0:
+    external: false
+  minio-0:
+    external: false
+  prometheus-0:
+    external: false
+name: risingwave-compose

integration_tests/mongodb/config-replica.js

@@ -0,0 +1,7 @@
+rsconf = {
+    _id: "rs0",
+    members: [{ _id: 0, host: "mongodb:27017", priority: 1.0 }],
+};
+rs.initiate(rsconf);
+rs.status();
+

integration_tests/mongodb/docker-compose.yaml

@@ -0,0 +1,39 @@
+version: "3"
+services:
+  mongodb:
+    image: mongo:4.4
+    container_name: mongodb
+    ports:
+      - "27017:27017"
+    command: --replSet rs0 --oplogSize 128
+    restart: always
+    healthcheck:
+      test: "echo 'db.runCommand({ping: 1})' | mongo"
+      interval: 5s
+      timeout: 10s
+      retries: 3
+  mongodb-setup:
+    image: mongo:4.4
+    container_name: mongodb-setup
+    depends_on:
+      - mongodb
+    entrypoint:
+      [
+        "bash",
+        "-c",
+        "sleep 10 && mongo --host mongodb:27017 /config-replica.js && sleep 10"
+      ]
+    restart: "no"
+    volumes:
+      - ./config-replica.js:/config-replica.js
+  random_data_generator:
+    build:
+      context: .
+      dockerfile: Dockerfile.generator
+    container_name: random_data_generator
+    depends_on:
+      - mongodb
+    environment:
+      MONGO_HOST: mongodb
+      MONGO_PORT: 27017
+      MONGO_DB_NAME: random_data

integration_tests/mongodb/requirements.txt

@@ -0,0 +1,2 @@
+pymongo
+Faker

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

@@ -0,0 +1,41 @@
+// Copyright 2024 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.common;
+
+public class CdcConnectorException extends RuntimeException {
+    private static final long serialVersionUID = 1L;
+
+    public CdcConnectorException() {}
+
+    public CdcConnectorException(String message) {
+        super(message);
+    }
+
+    public CdcConnectorException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public CdcConnectorException(Throwable cause) {
+        super(cause);
+    }
+
+    public CdcConnectorException(
+            String message,
+            Throwable cause,
+            boolean enableSuppression,
+            boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}