add integration test

Author: StrikeW

integration_tests/mongodb-cdc/Dockerfile.generator

@@ -0,0 +1,25 @@
+# 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 requirements.txt /app
+
+# Install any needed packages specified in requirements.txt
+RUN pip install -r requirements.txt
+
+# Copy the rest of the application code
+COPY . /app
+
+# 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"]

integration_tests/mongodb-cdc/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(1000):
+    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}")

integration_tests/mongodb-cdc/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-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,42 @@
+version: '3.1'
+
+services:
+  mongodb:
+    image: mongodb/mongodb-community-server:4.4.23-ubi8
+    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: mongodb/mongodb-community-server:4.4.23-ubi8
+    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-cdc/requirements.txt

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

java/connector-node/risingwave-source-test/src/test/java/com/risingwave/connector/source/MongoDbSourceTest.java

@@ -0,0 +1,104 @@
+// 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;
+
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.InsertManyOptions;
+import com.risingwave.connector.ConnectorServiceImpl;
+import com.risingwave.proto.ConnectorServiceProto;
+import io.grpc.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import org.bson.Document;
+import org.bson.types.ObjectId;
+import org.junit.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.MongoDBContainer;
+
+import static java.lang.Thread.sleep;
+
+public class MongoDbSourceTest {
+    private static final Logger LOG = LoggerFactory.getLogger(MongoDbSourceTest.class.getName());
+
+    public static Server connectorService =
+            ServerBuilder.forPort(SourceTestClient.DEFAULT_PORT)
+                    .addService(new ConnectorServiceImpl())
+                    .build();
+
+    public static SourceTestClient testClient =
+            new SourceTestClient(
+                    Grpc.newChannelBuilder(
+                                    "localhost:" + SourceTestClient.DEFAULT_PORT,
+                                    InsecureChannelCredentials.create())
+                            .build());
+
+    @BeforeClass
+    public static void init() {
+        try {
+            connectorService.start();
+        } catch (Exception e) {
+            LOG.error("failed to start connector service", e);
+            Assert.fail();
+        }
+    }
+
+    @AfterClass
+    public static void cleanup() {
+        connectorService.shutdown();
+    }
+
+    // manually test
+    @Test
+    @Ignore
+    public void testSnapshotLoad() throws Exception {
+        Map<String, String> props = new HashMap<>();
+        props.put("mongodb.url", "mongodb://localhost:27017/?replicaSet=rs0");
+        props.put("collection.name", "dev.*");
+        Iterator<ConnectorServiceProto.GetEventStreamResponse> eventStream =
+                testClient.getEventStream(ConnectorServiceProto.SourceType.MONGODB, 3001, props);
+        Callable<Integer> countTask =
+                () -> {
+                    int count = 0;
+                    while (eventStream.hasNext()) {
+                        List<ConnectorServiceProto.CdcMessage> messages =
+                                eventStream.next().getEventsList();
+                        for (ConnectorServiceProto.CdcMessage msg : messages) {
+                            if (!msg.getPayload().isBlank()) {
+                                count++;
+                            }
+                        }
+                        if (count >= 10) {
+                            return count;
+                        }
+                    }
+                    return count;
+                };
+
+        var pool = Executors.newFixedThreadPool(1);
+        var result = pool.submit(countTask);
+        Assert.assertEquals(10, result.get().intValue());
+    }
+}