This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new bccea91c92 [Fix][Connector-V2] Support MongoDB CDC SRV URI (#11129)
bccea91c92 is described below
commit bccea91c923c6fb4bcf4f52a3a71384f3b813f6a
Author: Jast <[email protected]>
AuthorDate: Mon Jun 22 20:55:26 2026 +0800
[Fix][Connector-V2] Support MongoDB CDC SRV URI (#11129)
---
docs/en/connectors/source/MongoDB-CDC.md | 4 +-
docs/zh/connectors/source/MongoDB-CDC.md | 4 +-
.../config/MongodbIncrementalSourceOptions.java | 4 +-
.../cdc/mongodb/utils/MongodbRecordUtils.java | 7 +-
.../seatunnel/cdc/mongodb/utils/MongodbUtils.java | 100 +++++++++++++++++++--
.../test/java/mongodb/utils/MongodbUtilsTest.java | 82 +++++++++++++++++
6 files changed, 187 insertions(+), 14 deletions(-)
diff --git a/docs/en/connectors/source/MongoDB-CDC.md
b/docs/en/connectors/source/MongoDB-CDC.md
index 3c153350fb..03d3bc221d 100644
--- a/docs/en/connectors/source/MongoDB-CDC.md
+++ b/docs/en/connectors/source/MongoDB-CDC.md
@@ -114,7 +114,7 @@ For specific types in MongoDB, we use Extended JSON format
to map them to Seatun
| Name | Type | Required | Default |
Description
|
|------------------------------------|--------|----------|---------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| hosts | String | Yes | - | The
comma-separated list of hostname and port pairs of the MongoDB servers. eg.
`localhost:27017,localhost:27018`
|
+| hosts | String | Yes | - | The
comma-separated list of hostname and port pairs of the MongoDB servers, or a
standard MongoDB connection URI using `mongodb://` or `mongodb+srv://`. eg.
`localhost:27017,localhost:27018` or `mongodb+srv://cluster.example.net`
|
| username | String | No | - | Name of
the database user to be used when connecting to MongoDB.
|
| password | String | No | - | Password
to be used when connecting to MongoDB.
|
| database | List | Yes | - | Name of
the database to watch for changes. If not set then all databases will be
captured. The database also supports regular expressions to monitor multiple
databases matching the regular expression. eg. `db1,db2`.
|
@@ -373,4 +373,4 @@ sink {
## Changelog
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/zh/connectors/source/MongoDB-CDC.md
b/docs/zh/connectors/source/MongoDB-CDC.md
index 83782ea380..9f0362608e 100644
--- a/docs/zh/connectors/source/MongoDB-CDC.md
+++ b/docs/zh/connectors/source/MongoDB-CDC.md
@@ -114,7 +114,7 @@ db.grantRolesToUser("<USER_NAME>", ["<ROLE_NAME>"])
| Name | 类型 | 必须 | 默认值 | 描述
|
|------------------------------------|--------|----------|-------|---------------------------------------------------------------------------------------|
-| hosts | String | 是 | - |
MongoDB服务器的主机名和端口对的逗号分隔列表。如 `localhost:27017,localhost:27018`
|
+| hosts | String | 是 | - |
MongoDB服务器的主机名和端口对的逗号分隔列表，也可以是使用 `mongodb://` 或 `mongodb+srv://` 的标准 MongoDB 连接
URI。如 `localhost:27017,localhost:27018` 或 `mongodb+srv://cluster.example.net`
|
| username | String | 否 | - |
连接到MongoDB时要使用的数据库用户的名称。
|
| password | String | 否 | - |
连接到MongoDB时使用的密码。
|
| database | List | 是 | - |
要监视更改的数据库的名称。如果未设置，则将捕获所有数据库。该数据库还支持正则表达式，以监视与正则表达式匹配的多个数据库。例如db1、db2。
|
@@ -373,4 +373,4 @@ sink {
## 修改日志
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbIncrementalSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbIncrementalSourceOptions.java
index ff2449e15d..2602449a7b 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbIncrementalSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbIncrementalSourceOptions.java
@@ -39,7 +39,9 @@ public class MongodbIncrementalSourceOptions extends
SourceOptions implements Ta
.noDefaultValue()
.withDescription(
"The comma-separated list of hostname and port
pairs of the MongoDB servers. "
- + "eg. localhost:27017,localhost:27018");
+ + "A standard MongoDB connection URI with
mongodb:// or mongodb+srv:// "
+ + "scheme is also supported. "
+ + "eg. localhost:27017,localhost:27018 or
mongodb+srv://cluster.example.net");
public static final Option<String> USERNAME =
Options.key("username")
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
index 1487962f20..9b378df424 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbRecordUtils.java
@@ -53,6 +53,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.Mongo
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.OUTPUT_SCHEMA;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.SOURCE_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils.buildConnectionNamespacePrefix;
public class MongodbRecordUtils {
@@ -194,9 +195,7 @@ public class MongodbRecordUtils {
public static @Nonnull Map<String, String> createPartitionMap(
String hosts, String database, String collection) {
- StringBuilder builder = new StringBuilder();
- builder.append("mongodb://");
- builder.append(hosts);
+ StringBuilder builder = new
StringBuilder(buildConnectionNamespacePrefix(hosts));
builder.append("/");
if (StringUtils.isNotEmpty(database)) {
builder.append(database);
@@ -209,7 +208,7 @@ public class MongodbRecordUtils {
}
public static @Nonnull Map<String, Object>
createHeartbeatPartitionMap(String hosts) {
- String builder = "mongodb://" + hosts + "/" + "__mongodb_heartbeats";
+ String builder = buildConnectionNamespacePrefix(hosts) + "/" +
"__mongodb_heartbeats";
return Collections.singletonMap(NS_FIELD, builder);
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
index 8dba927f0b..a446d748c4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/utils/MongodbUtils.java
@@ -84,6 +84,9 @@ import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.Collec
@Slf4j
public class MongodbUtils {
+ private static final String MONGODB_SCHEME = "mongodb://";
+ private static final String MONGODB_SRV_SCHEME = "mongodb+srv://";
+ private static final String DEFAULT_DATABASE = "/";
public static ChangeStreamDescriptor getChangeStreamDescriptor(
@Nonnull MongodbSourceConfig sourceConfig,
@@ -371,19 +374,106 @@ public class MongodbUtils {
public static @Nonnull ConnectionString buildConnectionString(
String username, String password, String hosts, String
connectionOptions) {
- StringBuilder sb = new StringBuilder("mongodb://");
+ String uri =
+ isConnectionUri(hosts)
+ ? buildFromUri(hosts, username, password,
connectionOptions)
+ : buildFromHosts(hosts, username, password,
connectionOptions);
+ return new ConnectionString(uri);
+ }
+ private static String buildFromHosts(
+ String hosts, String username, String password, String
connectionOptions) {
+ StringBuilder sb = new StringBuilder(MONGODB_SCHEME);
if (hasCredentials(username, password)) {
appendCredentials(sb, username, password);
}
-
sb.append(hosts);
-
if (StringUtils.isNotEmpty(connectionOptions)) {
sb.append("/?").append(connectionOptions);
}
+ return sb.toString();
+ }
+
+ private static String buildFromUri(
+ String uri, String username, String password, String
connectionOptions) {
+ StringBuilder sb = new StringBuilder(uri);
+ if (hasCredentials(username, password) && !hasCredentialsInUri(uri)) {
+ sb.insert(getScheme(uri).length(), credentialString(username,
password));
+ }
+ appendConnectionOptions(sb, connectionOptions);
+ return sb.toString();
+ }
+
+ public static boolean isConnectionUri(String hosts) {
+ return StringUtils.startsWith(hosts, MONGODB_SCHEME)
+ || StringUtils.startsWith(hosts, MONGODB_SRV_SCHEME);
+ }
+
+ public static String buildConnectionNamespacePrefix(String hosts) {
+ if (!isConnectionUri(hosts)) {
+ return MONGODB_SCHEME + hosts;
+ }
+
+ String scheme = getScheme(hosts);
+ String remaining = hosts.substring(scheme.length());
+ int boundary = findAuthorityBoundary(remaining);
+ String authority = boundary == -1 ? remaining : remaining.substring(0,
boundary);
+ int credentialIndex = authority.lastIndexOf('@');
+ if (credentialIndex >= 0) {
+ authority = authority.substring(credentialIndex + 1);
+ }
+ return scheme + authority;
+ }
+
+ private static String getScheme(String uri) {
+ return StringUtils.startsWith(uri, MONGODB_SRV_SCHEME)
+ ? MONGODB_SRV_SCHEME
+ : MONGODB_SCHEME;
+ }
+
+ private static boolean hasCredentialsInUri(String uri) {
+ String scheme = getScheme(uri);
+ String remaining = uri.substring(scheme.length());
+ int boundary = findAuthorityBoundary(remaining);
+ String authority = boundary == -1 ? remaining : remaining.substring(0,
boundary);
+ return authority.contains("@");
+ }
+
+ private static int findAuthorityBoundary(String uriWithoutScheme) {
+ int slashIndex = uriWithoutScheme.indexOf('/');
+ int queryIndex = uriWithoutScheme.indexOf('?');
+ if (slashIndex == -1) {
+ return queryIndex;
+ }
+ if (queryIndex == -1) {
+ return slashIndex;
+ }
+ return Math.min(slashIndex, queryIndex);
+ }
+
+ private static String credentialString(String username, String password) {
+ return encodeValue(username) + ":" + encodeValue(password) + "@";
+ }
+
+ private static void appendConnectionOptions(StringBuilder sb, String
connectionOptions) {
+ if (StringUtils.isEmpty(connectionOptions)) {
+ return;
+ }
+ if (StringUtils.contains(sb, "?")) {
+ if (!StringUtils.endsWith(sb, "?") && !StringUtils.endsWith(sb,
"&")) {
+ sb.append("&");
+ }
+ } else if (hasDatabasePath(sb)) {
+ sb.append("?");
+ } else {
+ sb.append(DEFAULT_DATABASE).append("?");
+ }
+ sb.append(connectionOptions);
+ }
- return new ConnectionString(sb.toString());
+ private static boolean hasDatabasePath(StringBuilder sb) {
+ int schemeEnd = sb.indexOf("://") + 3;
+ return sb.indexOf("/", schemeEnd) >= 0;
}
private static boolean hasCredentials(String username, String password) {
@@ -392,7 +482,7 @@ public class MongodbUtils {
private static void appendCredentials(
@Nonnull StringBuilder sb, String username, String password) {
-
sb.append(encodeValue(username)).append(":").append(encodeValue(password)).append("@");
+ sb.append(credentialString(username, password));
}
public static String encodeValue(String value) {
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
new file mode 100644
index 0000000000..b441717da6
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/utils/MongodbUtilsTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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 mongodb.utils;
+
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbRecordUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConstants.NS_FIELD;
+
+public class MongodbUtilsTest {
+
+ @Test
+ public void testBuildConnectionStringFromHostsKeepsLegacyFormat() {
+ String connectionString =
+ MongodbUtils.buildConnectionString(
+ "user", "password", "localhost:27017",
"replicaSet=test")
+ .getConnectionString();
+
+ Assertions.assertEquals(
+ "mongodb://user:password@localhost:27017/?replicaSet=test",
connectionString);
+ }
+
+ @Test
+ public void testBuildConnectionStringKeepsSrvUri() {
+ String srvUri =
+ "mongodb+srv://user:[email protected]/test"
+ + "?retryWrites=true&w=majority";
+
+ String connectionString =
+ MongodbUtils.buildConnectionString(null, null, srvUri,
null).getConnectionString();
+
+ Assertions.assertEquals(srvUri, connectionString);
+ }
+
+ @Test
+ public void testBuildConnectionStringAddsCredentialsAndOptionsToSrvUri() {
+ String connectionString =
+ MongodbUtils.buildConnectionString(
+ "user",
+ "password",
+ "mongodb+srv://cluster0.example.mongodb.net",
+ "retryWrites=true")
+ .getConnectionString();
+
+ Assertions.assertEquals(
+
"mongodb+srv://user:[email protected]/?retryWrites=true",
+ connectionString);
+ }
+
+ @Test
+ public void testPartitionMapUsesSrvNamespaceWithoutCredentialsOrOptions() {
+ String hosts =
+
"mongodb+srv://user:[email protected]/admin"
+ + "?retryWrites=true";
+
+ Assertions.assertEquals(
+
"mongodb+srv://cluster0.example.mongodb.net/inventory.products",
+ MongodbRecordUtils.createPartitionMap(hosts, "inventory",
"products")
+ .get(NS_FIELD));
+ Assertions.assertEquals(
+
"mongodb+srv://cluster0.example.mongodb.net/__mongodb_heartbeats",
+
MongodbRecordUtils.createHeartbeatPartitionMap(hosts).get(NS_FIELD));
+ }
+}
