This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 54dec83e12 [INLONG-8959][Sort] Add Mongodb-cdc-connector on flink1.15
(#9011)
54dec83e12 is described below
commit 54dec83e1202466eec7628b0dc7b81cfe8798237
Author: Curry-CPU <[email protected]>
AuthorDate: Fri Oct 13 16:28:37 2023 +0800
[INLONG-8959][Sort] Add Mongodb-cdc-connector on flink1.15 (#9011)
---
.../src/main/assemblies/sort-connectors-v1.15.xml | 8 +
inlong-sort/sort-core/pom.xml | 6 +
.../sort-end-to-end-tests-v1.15/pom.xml | 20 ++
.../inlong/sort/tests/MongodbToStarRocksTest.java | 203 +++++++++++++++++++++
.../apache/inlong/sort/tests/MysqlToRocksTest.java | 3 +-
.../sort/tests/SqlserverToStarRocksTest.java | 2 +-
.../src/test/resources/docker/mongodb/setup.sql | 31 ++++
.../src/test/resources/flinkSql/mongodb_test.sql | 31 ++++
.../sort-connectors/mongodb-cdc/pom.xml | 155 ++++++++++++++++
.../inlong/sort/mongodb/MongoDBTableFactory.java | 152 +++++++++++++++
.../org.apache.flink.table.factories.Factory | 16 ++
.../sort-flink-v1.15/sort-connectors/pom.xml | 1 +
12 files changed, 625 insertions(+), 3 deletions(-)
diff --git a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
index b7fead209a..90dfa48893 100644
--- a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
+++ b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
@@ -51,6 +51,14 @@
</includes>
<fileMode>0644</fileMode>
</fileSet>
+ <fileSet>
+
<directory>../inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/target</directory>
+ <outputDirectory>inlong-sort/connectors</outputDirectory>
+ <includes>
+
<include>sort-connector-mongodb-cdc-v1.15-${project.version}.jar</include>
+ </includes>
+ <fileMode>0644</fileMode>
+ </fileSet>
<fileSet>
<directory>../inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/sqlserver-cdc/target</directory>
<outputDirectory>inlong-sort/connectors</outputDirectory>
diff --git a/inlong-sort/sort-core/pom.xml b/inlong-sort/sort-core/pom.xml
index 8e78f08a1c..50b41d6833 100644
--- a/inlong-sort/sort-core/pom.xml
+++ b/inlong-sort/sort-core/pom.xml
@@ -257,6 +257,12 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>sort-connector-mongodb-cdc-v1.15</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>sort-connector-sqlserver-cdc-v1.15</artifactId>
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/pom.xml
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/pom.xml
index 37244cf14f..150a3d2ab7 100644
--- a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/pom.xml
+++ b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/pom.xml
@@ -31,6 +31,7 @@
<properties>
<inlong.root.dir>${project.parent.parent.parent.basedir}</inlong.root.dir>
<flink.version>1.15.4</flink.version>
+ <mongodb-driver-sync.version>4.10.2</mongodb-driver-sync.version>
</properties>
<dependencies>
@@ -47,6 +48,17 @@
<artifactId>postgresql</artifactId>
<version>${testcontainers.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>mongodb</artifactId>
+ <version>${testcontainers.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.mongodb</groupId>
+ <artifactId>mongodb-driver-sync</artifactId>
+ <version>${mongodb-driver-sync.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
@@ -179,6 +191,14 @@
<type>jar</type>
<outputDirectory>${project.build.directory}/dependencies</outputDirectory>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.inlong</groupId>
+
<artifactId>sort-connector-mongodb-cdc-v1.15</artifactId>
+ <version>${project.version}</version>
+
<destFileName>sort-connector-mongodb-cdc.jar</destFileName>
+ <type>jar</type>
+
<outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.inlong</groupId>
<artifactId>sort-connector-mysql-cdc-v1.15</artifactId>
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MongodbToStarRocksTest.java
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MongodbToStarRocksTest.java
new file mode 100644
index 0000000000..673077308d
--- /dev/null
+++
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MongodbToStarRocksTest.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.sort.tests;
+
+import org.apache.inlong.sort.tests.utils.FlinkContainerTestEnv;
+import org.apache.inlong.sort.tests.utils.JdbcProxy;
+import org.apache.inlong.sort.tests.utils.StarRocksContainer;
+import org.apache.inlong.sort.tests.utils.TestUtils;
+
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoDatabase;
+import org.bson.Document;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.MongoDBContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
+
+import java.net.URISyntaxException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static com.mongodb.client.model.Filters.eq;
+import static com.mongodb.client.model.Updates.*;
+
+/**
+ * End-to-end tests for sort-connector-mongodb-cdc-v1.15 uber jar.
+ * Test flink sql Mongodb cdc to StarRocks
+ */
+public class MongodbToStarRocksTest extends FlinkContainerTestEnv {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MongodbToStarRocksTest.class);
+
+ private static final Path mongodbJar =
TestUtils.getResource("sort-connector-mongodb-cdc.jar");
+ private static final Path jdbcJar =
TestUtils.getResource("sort-connector-starrocks.jar");
+ private static final Path mysqlJdbcJar =
TestUtils.getResource("mysql-driver.jar");
+
+ private static final Logger STAR_ROCKS_LOG =
LoggerFactory.getLogger(StarRocksContainer.class);
+
+ private static final String sqlFile;
+
+ //
----------------------------------------------------------------------------------------
+ // StarRocks Variables
+ //
----------------------------------------------------------------------------------------
+ private static final String INTER_CONTAINER_STAR_ROCKS_ALIAS = "starrocks";
+ private static final String NEW_STARROCKS_REPOSITORY = "inlong-starrocks";
+ private static final String NEW_STARROCKS_TAG = "latest";
+ private static final String STAR_ROCKS_IMAGE_NAME =
"starrocks/allin1-ubi:3.0.4";
+
+ static {
+ try {
+ sqlFile =
Paths.get(PostgresToStarRocksTest.class.getResource("/flinkSql/mongodb_test.sql").toURI())
+ .toString();
+ buildStarRocksImage();
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static String getNewStarRocksImageName() {
+ return NEW_STARROCKS_REPOSITORY + ":" + NEW_STARROCKS_TAG;
+ }
+
+ public static void buildStarRocksImage() {
+ GenericContainer oldStarRocks = new
GenericContainer(STAR_ROCKS_IMAGE_NAME);
+ Startables.deepStart(Stream.of(oldStarRocks)).join();
+
oldStarRocks.copyFileToContainer(MountableFile.forClasspathResource("/docker/starrocks/start_fe_be.sh"),
+ "/data/deploy/");
+ try {
+ oldStarRocks.execInContainer("chmod", "+x",
"/data/deploy/start_fe_be.sh");
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ oldStarRocks.getDockerClient()
+ .commitCmd(oldStarRocks.getContainerId())
+ .withRepository(NEW_STARROCKS_REPOSITORY)
+ .withTag(NEW_STARROCKS_TAG).exec();
+ oldStarRocks.stop();
+ }
+
+ @ClassRule
+ public static StarRocksContainer STAR_ROCKS =
+ (StarRocksContainer) new
StarRocksContainer(getNewStarRocksImageName())
+ .withExposedPorts(9030, 8030, 8040)
+ .withNetwork(NETWORK)
+ .withAccessToHost(true)
+ .withNetworkAliases(INTER_CONTAINER_STAR_ROCKS_ALIAS)
+ .withLogConsumer(new Slf4jLogConsumer(STAR_ROCKS_LOG));
+
+ @ClassRule
+ public static MongoDBContainer MONGODB_CONTAINER = new MongoDBContainer(
+
DockerImageName.parse("mongo:4.0.10").asCompatibleSubstituteFor("mongo"))
+ .withNetwork(NETWORK)
+ .withNetworkAliases("mongo")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+
+ @Before
+ public void setup() {
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ initializeStarRocksTable();
+ }
+
+ private void initializeStarRocksTable() {
+ try (Connection conn =
+ DriverManager.getConnection(STAR_ROCKS.getJdbcUrl(),
STAR_ROCKS.getUsername(),
+ STAR_ROCKS.getPassword());
+ Statement stat = conn.createStatement()) {
+ stat.execute("CREATE TABLE IF NOT EXISTS test_output1 (\n"
+ + " _id INT NOT NULL,\n"
+ + " name VARCHAR(255) NOT NULL DEFAULT 'flink',\n"
+ + " description VARCHAR(512)\n"
+ + ")\n"
+ + "PRIMARY KEY(_id)\n"
+ + "DISTRIBUTED by HASH(_id) PROPERTIES
(\"replication_num\" = \"1\");");
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @AfterClass
+ public static void teardown() {
+ if (MONGODB_CONTAINER != null) {
+ MONGODB_CONTAINER.stop();
+ }
+ if (STAR_ROCKS != null) {
+ STAR_ROCKS.stop();
+ }
+ }
+
+ /**
+ * Test flink sql postgresql cdc to StarRocks
+ *
+ * @throws Exception The exception may throws when execute the case
+ */
+ @Test
+ public void testMongodbUpdateAndDelete() throws Exception {
+ submitSQLJob(sqlFile, jdbcJar, mongodbJar, mysqlJdbcJar);
+ waitUntilJobRunning(Duration.ofSeconds(10));
+
+ // generate input
+ MongoClient mongoClient =
MongoClients.create(MONGODB_CONTAINER.getConnectionString());
+ MongoDatabase database = mongoClient.getDatabase("test");
+ MongoCollection<Document> collection =
database.getCollection("test_input1");
+
+ Document document1 = new Document("_id", 1)
+ .append("name", "jacket")
+ .append("description", "water resistent white wind breaker");
+ Document document2 = new Document("_id", 2)
+ .append("name", "scooter")
+ .append("description", "Big 2-wheel scooter ");
+ List<Document> documents = new ArrayList<Document>();
+ documents.add(document1);
+ documents.add(document2);
+ collection.insertMany(documents);
+ collection.updateOne(eq("_id", 2), combine(set("name", "tom")));
+ collection.deleteOne(eq("_id", 1));
+
+ JdbcProxy proxy =
+ new JdbcProxy(STAR_ROCKS.getJdbcUrl(),
STAR_ROCKS.getUsername(),
+ STAR_ROCKS.getPassword(),
+ STAR_ROCKS.getDriverClassName());
+ List<String> expectResult =
+ Arrays.asList("2,tom,Big 2-wheel scooter ");
+ proxy.checkResultWithTimeout(
+ expectResult,
+ "test_output1",
+ 3,
+ 60000L);
+ }
+}
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MysqlToRocksTest.java
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MysqlToRocksTest.java
index 0b777cf429..e0415e24f0 100644
---
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MysqlToRocksTest.java
+++
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/MysqlToRocksTest.java
@@ -21,7 +21,6 @@ import
org.apache.inlong.sort.tests.utils.FlinkContainerTestEnv;
import org.apache.inlong.sort.tests.utils.JdbcProxy;
import org.apache.inlong.sort.tests.utils.MySqlContainer;
import org.apache.inlong.sort.tests.utils.StarRocksContainer;
-import org.apache.inlong.sort.tests.utils.StarRocksManager;
import org.apache.inlong.sort.tests.utils.TestUtils;
import org.junit.AfterClass;
@@ -45,9 +44,9 @@ import java.util.List;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.INTER_CONTAINER_STAR_ROCKS_ALIAS;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.STAR_ROCKS_LOG;
+import static
org.apache.inlong.sort.tests.utils.StarRocksManager.buildStarRocksImage;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.getNewStarRocksImageName;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.initializeStarRocksTable;
-import static
org.apache.inlong.sort.tests.utils.StarRocksManager.buildStarRocksImage;
/**
* End-to-end tests for sort-connector-postgres-cdc-v1.15 uber jar.
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/SqlserverToStarRocksTest.java
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/SqlserverToStarRocksTest.java
index 01ab9eadb1..194fb2cfd5 100644
---
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/SqlserverToStarRocksTest.java
+++
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/SqlserverToStarRocksTest.java
@@ -45,9 +45,9 @@ import java.util.List;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.INTER_CONTAINER_STAR_ROCKS_ALIAS;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.STAR_ROCKS_LOG;
+import static
org.apache.inlong.sort.tests.utils.StarRocksManager.buildStarRocksImage;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.getNewStarRocksImageName;
import static
org.apache.inlong.sort.tests.utils.StarRocksManager.initializeStarRocksTable;
-import static
org.apache.inlong.sort.tests.utils.StarRocksManager.buildStarRocksImage;
public class SqlserverToStarRocksTest extends FlinkContainerTestEnv {
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/docker/mongodb/setup.sql
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/docker/mongodb/setup.sql
new file mode 100644
index 0000000000..75444ae4fe
--- /dev/null
+++
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/docker/mongodb/setup.sql
@@ -0,0 +1,31 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you 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.
+
+-- In production you would almost certainly limit the replication user must be
on the follower (slave) machine,
+-- to prevent other clients accessing the log from other machines. For
example, 'replicator'@'follower.acme.com'.
+-- However, in this database we'll grant 2 users different privileges:
+--
+-- 1) 'flinkuser' - all privileges required by the snapshot reader AND binlog
reader (used for testing)
+-- 2) 'inlong' - all privileges
+--
+use admin;
+db.createUser({
+ user: "flinkuser",
+ pwd: "flinkpw",
+ roles: [
+ { role: "read", db: "inventory" },
+ { role: "readAnyDatabase", db: "inventory" }
+ ]
+});
\ No newline at end of file
diff --git
a/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/flinkSql/mongodb_test.sql
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/flinkSql/mongodb_test.sql
new file mode 100644
index 0000000000..ad658b924b
--- /dev/null
+++
b/inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/flinkSql/mongodb_test.sql
@@ -0,0 +1,31 @@
+SET 'execution.checkpointing.interval' = '3s';
+CREATE TABLE test_input1 (
+ _id INT primary key,
+ name STRING,
+ description STRING
+) WITH (
+ 'connector' = 'mongodb-cdc-inlong',
+ 'hosts' = 'mongo:27017',
+ 'database' = 'test',
+ 'collection' = 'test_input1',
+ 'connection.options' = 'connectTimeoutMS=30000&maxIdleTimeMS=20000'
+);
+
+CREATE TABLE test_output1 (
+ _id INT primary key,
+ name STRING,
+ description STRING
+) WITH (
+ 'connector' = 'starrocks-inlong',
+ 'jdbc-url' = 'jdbc:mysql://starrocks:9030',
+ 'load-url'='starrocks:8030',
+ 'database-name'='test',
+ 'table-name' = 'test_output1',
+ 'username' = 'inlong',
+ 'password' = 'inlong',
+ 'sink.properties.format' = 'json',
+ 'sink.properties.strip_outer_array' = 'true',
+ 'sink.buffer-flush.interval-ms' = '1000'
+);
+
+INSERT INTO test_output1 select * from test_input1;
\ No newline at end of file
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/pom.xml
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/pom.xml
new file mode 100644
index 0000000000..60357dde2e
--- /dev/null
+++
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/pom.xml
@@ -0,0 +1,155 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>sort-connectors-v1.15</artifactId>
+ <version>1.10.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>sort-connector-mongodb-cdc-v1.15</artifactId>
+ <packaging>jar</packaging>
+ <name>Apache InLong - Sort-connector-mongodb-cdc</name>
+
+ <properties>
+
<inlong.root.dir>${project.parent.parent.parent.parent.parent.basedir}</inlong.root.dir>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-connector-mongodb-cdc</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.debezium</groupId>
+ <artifactId>debezium-api</artifactId>
+ <version>${debezium.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.debezium</groupId>
+ <artifactId>debezium-embedded</artifactId>
+ <version>${debezium.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.debezium</groupId>
+ <artifactId>debezium-core</artifactId>
+ <version>${debezium.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>sort-connector-cdc-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>sort-connector-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <artifactSet>
+ <includes>
+ <include>org.apache.inlong:*</include>
+ <include>io.debezium:debezium-api</include>
+
<include>io.debezium:debezium-embedded</include>
+
<include>io.debezium:debezium-core</include>
+
<include>com.ververica:flink-connector-debezium</include>
+
<include>com.ververica:flink-connector-mongodb-cdc</include>
+
<include>org.mongodb.kafka:mongo-kafka-connect</include>
+
<include>org.mongodb:mongodb-driver-sync</include>
+
<include>org.mongodb:mongodb-driver-core</include>
+ <include>org.mongodb:bson</include>
+ <include>org.apache.avro:avro</include>
+ <include>org.apache.kafka:*</include>
+ <include>com.fasterxml.*:*</include>
+ <include>com.google.code.gson:*</include>
+ <include>com.google.guava:*</include>
+ <!-- Include fixed version 18.0-13.0 of
flink shaded guava -->
+
<include>org.apache.flink:flink-shaded-guava</include>
+ <include>com.google.protobuf:*</include>
+
<include>com.ververica:flink-cdc-base</include>
+ </includes>
+ </artifactSet>
+ <filters>
+ <filter>
+
<artifact>org.apache.inlong:sort-connector-*</artifact>
+ <includes>
+ <include>org/apache/inlong/**</include>
+
<include>META-INF/services/org.apache.flink.table.factories.Factory</include>
+ </includes>
+ </filter>
+ <filter>
+ <artifact>org.apache.kafka:*</artifact>
+ <excludes>
+
<exclude>kafka/kafka-version.properties</exclude>
+ <exclude>LICENSE</exclude>
+ <!-- Does not contain anything
relevant.
+ Cites a binary dependency on
jersey, but this is neither reflected in the
+ dependency graph, nor are any
jersey files bundled. -->
+ <exclude>NOTICE</exclude>
+ <exclude>common/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <relocation>
+
<pattern>org.apache.inlong.sort.base</pattern>
+
<shadedPattern>org.apache.inlong.sort.cdc.mongodb.shaded.org.apache.inlong.sort.base</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.kafka</pattern>
+
<shadedPattern>org.apache.inlong.sort.cdc.mongodb.shaded.org.apache.kafka</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.avro</pattern>
+
<shadedPattern>org.apache.inlong.sort.cdc.mongodb.shaded.org.apache.avro</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.fasterxml</pattern>
+
<shadedPattern>org.apache.inlong.sort.cdc.mongodb.shaded.com.fasterxml</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google</pattern>
+
<shadedPattern>org.apache.inlong.sort.cdc.mongodb.shaded.com.google</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/mongodb/MongoDBTableFactory.java
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/mongodb/MongoDBTableFactory.java
new file mode 100644
index 0000000000..308f1179ef
--- /dev/null
+++
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/mongodb/MongoDBTableFactory.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.sort.mongodb;
+
+import com.ververica.cdc.connectors.mongodb.table.MongoDBTableSource;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import java.time.ZoneId;
+import java.util.HashSet;
+import java.util.Set;
+
+import static
com.ververica.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
+import static
com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.*;
+import static
com.ververica.cdc.debezium.utils.ResolvedSchemaUtils.getPhysicalSchema;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.inlong.sort.base.Constants.AUDIT_KEYS;
+import static org.apache.inlong.sort.base.Constants.INLONG_AUDIT;
+import static org.apache.inlong.sort.base.Constants.INLONG_METRIC;
+import static org.apache.inlong.sort.base.Constants.SOURCE_MULTIPLE_ENABLE;
+
+public class MongoDBTableFactory implements DynamicTableSourceFactory {
+
+ private static final String IDENTIFIER = "mongodb-cdc-inlong";
+
+ private static final String DOCUMENT_ID_FIELD = "_id";
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ final FactoryUtil.TableFactoryHelper helper =
+ FactoryUtil.createTableFactoryHelper(this, context);
+ helper.validate();
+
+ final ReadableConfig config = helper.getOptions();
+
+ String hosts = config.get(HOSTS);
+ String connectionOptions =
config.getOptional(CONNECTION_OPTIONS).orElse(null);
+
+ String username = config.getOptional(USERNAME).orElse(null);
+ String password = config.getOptional(PASSWORD).orElse(null);
+
+ String database = config.getOptional(DATABASE).orElse(null);
+ String collection = config.getOptional(COLLECTION).orElse(null);
+
+ Integer batchSize = config.get(BATCH_SIZE);
+ Integer pollMaxBatchSize = config.get(POLL_MAX_BATCH_SIZE);
+ Integer pollAwaitTimeMillis = config.get(POLL_AWAIT_TIME_MILLIS);
+
+ Integer heartbeatIntervalMillis =
config.get(HEARTBEAT_INTERVAL_MILLIS);
+
+ Boolean copyExisting = config.get(COPY_EXISTING);
+ Integer copyExistingQueueSize =
config.getOptional(COPY_EXISTING_QUEUE_SIZE).orElse(null);
+
+ String zoneId =
context.getConfiguration().get(TableConfigOptions.LOCAL_TIME_ZONE);
+ ZoneId localTimeZone =
+
TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zoneId)
+ ? ZoneId.systemDefault()
+ : ZoneId.of(zoneId);
+
+ boolean enableParallelRead =
config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
+
+ int splitSizeMB = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB);
+ int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);
+
+ ResolvedSchema physicalSchema =
+
getPhysicalSchema(context.getCatalogTable().getResolvedSchema());
+ checkArgument(physicalSchema.getPrimaryKey().isPresent(), "Primary key
must be present");
+ checkPrimaryKey(physicalSchema.getPrimaryKey().get(), "Primary key
must be _id field");
+
+ return new MongoDBTableSource(
+ physicalSchema,
+ hosts,
+ username,
+ password,
+ database,
+ collection,
+ connectionOptions,
+ copyExisting,
+ copyExistingQueueSize,
+ batchSize,
+ pollMaxBatchSize,
+ pollAwaitTimeMillis,
+ heartbeatIntervalMillis,
+ localTimeZone,
+ enableParallelRead,
+ splitMetaGroupSize,
+ splitSizeMB);
+ }
+
+ private void checkPrimaryKey(UniqueConstraint pk, String message) {
+ checkArgument(
+ pk.getColumns().size() == 1 &&
pk.getColumns().contains(DOCUMENT_ID_FIELD),
+ message);
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(HOSTS);
+ options.add(DATABASE);
+ options.add(COLLECTION);
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(USERNAME);
+ options.add(PASSWORD);
+ options.add(CONNECTION_OPTIONS);
+ options.add(COPY_EXISTING);
+ options.add(COPY_EXISTING_QUEUE_SIZE);
+ options.add(BATCH_SIZE);
+ options.add(POLL_MAX_BATCH_SIZE);
+ options.add(POLL_AWAIT_TIME_MILLIS);
+ options.add(HEARTBEAT_INTERVAL_MILLIS);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB);
+ options.add(CHUNK_META_GROUP_SIZE);
+ options.add(AUDIT_KEYS);
+ options.add(INLONG_METRIC);
+ options.add(INLONG_AUDIT);
+ options.add(SOURCE_MULTIPLE_ENABLE);
+ return options;
+ }
+
+}
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000000..a9362ccdb2
--- /dev/null
+++
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+org.apache.inlong.sort.mongodb.MongoDBTableFactory
\ No newline at end of file
diff --git a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml
index 74c17313ff..82a4b2949e 100644
--- a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml
+++ b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pom.xml
@@ -37,6 +37,7 @@
<module>sqlserver-cdc</module>
<module>mysql-cdc</module>
<module>iceberg</module>
+ <module>mongodb-cdc</module>
<module>pulsar</module>
</modules>
