This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 4007c735f [FLINK-39610] Mongo CDC supports specifying individual SSL
context (#4393)
4007c735f is described below
commit 4007c735fffab4279f4def67cc6705bcfbd88efa
Author: yuxiqian <[email protected]>
AuthorDate: Wed Sep 23 10:26:05 2026 +0800
[FLINK-39610] Mongo CDC supports specifying individual SSL context (#4393)
---
.../docs/connectors/flink-sources/mongodb-cdc.md | 58 +++
.../docs/connectors/flink-sources/mongodb-cdc.md | 58 +++
.../mongodb/source/MongoDBSourceBuilder.java | 48 +++
.../mongodb/source/config/MongoDBSourceConfig.java | 83 ++++-
.../source/config/MongoDBSourceConfigFactory.java | 69 +++-
.../source/config/MongoDBSourceOptions.java | 64 ++++
.../mongodb/source/connection/MongoClientPool.java | 44 ++-
.../mongodb/source/utils/MongoDBSslUtils.java | 121 ++++++
.../mongodb/table/MongoDBTableSource.java | 67 +++-
.../mongodb/table/MongoDBTableSourceFactory.java | 35 +-
.../mongodb/source/MongoDBSslConnectionITCase.java | 410 +++++++++++++++++++++
.../mongodb/source/utils/MongoDBSslUtilsTest.java | 294 +++++++++++++++
.../mongodb/table/MongoDBTableFactoryTest.java | 69 +++-
.../mongodb/utils/MongoDBSslContainer.java | 179 +++++++++
.../src/test/resources/ssl/ca.crt | 20 +
.../src/test/resources/ssl/ca.key | 28 ++
.../src/test/resources/ssl/ca.srl | 1 +
.../src/test/resources/ssl/client-keystore.jks | Bin 0 -> 2197 bytes
.../src/test/resources/ssl/client-keystore.p12 | Bin 0 -> 2690 bytes
.../src/test/resources/ssl/client-truststore.jks | Bin 0 -> 917 bytes
.../src/test/resources/ssl/client-truststore.p12 | Bin 0 -> 1222 bytes
.../src/test/resources/ssl/client.crt | 20 +
.../src/test/resources/ssl/client.csr | 16 +
.../src/test/resources/ssl/client.key | 28 ++
.../src/test/resources/ssl/client.pem | 48 +++
.../src/test/resources/ssl/server.crt | 21 ++
.../src/test/resources/ssl/server.csr | 16 +
.../src/test/resources/ssl/server.ext | 2 +
.../src/test/resources/ssl/server.key | 28 ++
.../src/test/resources/ssl/server.pem | 49 +++
pom.xml | 6 +
31 files changed, 1863 insertions(+), 19 deletions(-)
diff --git a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
index 83dda0ecf..02a0cb345 100644
--- a/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content.zh/docs/connectors/flink-sources/mongodb-cdc.md
@@ -354,6 +354,62 @@ MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能
For example updating an already updated value in snapshot, or deleting
an already deleted entry in snapshot. These replayed change log events should
be handled specially.
</td>
</tr>
+ <tr>
+ <td>mongodb.ssl.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>连接器是否使用 SSL 连接 MongoDB 实例。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.invalid.hostname.allowed</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>启用 SSL 时，该配置控制是否在连接阶段禁用严格的主机名校验。若设为
<code>true</code>，连接将无法防范中间人攻击。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>密钥库（keystore）文件的位置。该项为可选，可用于客户端与 MongoDB 服务端之间的双向认证。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore.password</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>密钥库文件的密码。该项为可选，仅在配置了 <code>mongodb.ssl.keystore</code> 时才需要。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore.type</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">PKCS12</td>
+ <td>String</td>
+ <td>密钥库文件的类型。该项为可选，仅在配置了 <code>mongodb.ssl.keystore</code> 时才需要。默认为
PKCS12。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>用于校验服务端证书的信任库（truststore）文件的位置。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore.password</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>信任库文件的密码。用于校验信任库的完整性并解锁信任库。</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore.type</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">PKCS12</td>
+ <td>String</td>
+ <td>信任库文件的类型。该项为可选，仅在配置了 <code>mongodb.ssl.truststore</code> 时才需要。默认为
PKCS12。</td>
+ </tr>
</tbody>
</table>
</div>
@@ -361,6 +417,8 @@ MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能
注意: `heartbeat.interval.ms` 强烈建议设置一个大于 0 的适当值 **如果集合更改缓慢**.
当我们从检查点或保存点恢复 Flink 作业时，心跳事件可以向前推送 `resumeToken`，以避免 `resumeToken` 过期。
+注意: `mongodb.ssl.*` 相关配置仅在开启增量快照（`scan.incremental.snapshot.enabled` =
`true`）时才会生效，非增量快照的数据源会忽略这些配置。
+
可用元数据
----------------
diff --git a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
index 5eed10d75..f1c0fc5af 100644
--- a/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
+++ b/docs/content/docs/connectors/flink-sources/mongodb-cdc.md
@@ -379,6 +379,62 @@ Connector Options
For example updating an already updated value in snapshot, or deleting
an already deleted entry in snapshot. These replayed change log events should
be handled specially.
</td>
</tr>
+ <tr>
+ <td>mongodb.ssl.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether the connector will use SSL to connect to MongoDB
instances.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.invalid.hostname.allowed</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>When SSL is enabled, this setting controls whether strict hostname
checking is disabled during the connection phase. If <code>true</code>, the
connection will not prevent man-in-the-middle attacks.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The location of the key store file. This is optional and can be used
for two-way authentication between the client and the MongoDB server.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore.password</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The password for the key store file. This is optional and only
needed if <code>mongodb.ssl.keystore</code> is configured.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.keystore.type</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">PKCS12</td>
+ <td>String</td>
+ <td>The type of key store file. This is optional and only needed if
<code>mongodb.ssl.keystore</code> is configured. Defaults to PKCS12.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The location of the trust store file for the server certificate
verification.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore.password</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The password for the trust store file. Used to check the integrity
of the truststore, and unlock the truststore.</td>
+ </tr>
+ <tr>
+ <td>mongodb.ssl.truststore.type</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">PKCS12</td>
+ <td>String</td>
+ <td>The type of trust store file. This is optional and only needed if
<code>mongodb.ssl.truststore</code> is configured. Defaults to PKCS12.</td>
+ </tr>
</tbody>
</table>
</div>
@@ -386,6 +442,8 @@ Connector Options
Note: `heartbeat.interval.ms` is highly recommended setting a proper value
larger than 0 **if the collection changes slowly**.
The heartbeat event can push the `resumeToken` forward to avoid `resumeToken`
being expired when we recover the Flink job from a checkpoint or savepoint.
+Note: The `mongodb.ssl.*` options only take effect when the incremental
snapshot is enabled (`scan.incremental.snapshot.enabled` = `true`). They are
ignored by the non-incremental source.
+
Available Metadata
----------------
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
index 553a533fd..365ed7fe7 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java
@@ -295,4 +295,52 @@ public class MongoDBSourceBuilder<T> {
public MongoDBSource<T> build() {
return new MongoDBSource<>(configFactory, checkNotNull(deserializer));
}
+
+ /** Whether the connector will use SSL to connect to MongoDB instances. */
+ public MongoDBSourceBuilder<T> sslEnabled(boolean sslEnabled) {
+ this.configFactory.sslEnabled(sslEnabled);
+ return this;
+ }
+
+ /** When SSL is enabled, controls whether strict hostname checking is
disabled. */
+ public MongoDBSourceBuilder<T> sslInvalidHostnameAllowed(boolean
sslInvalidHostnameAllowed) {
+
this.configFactory.sslInvalidHostnameAllowed(sslInvalidHostnameAllowed);
+ return this;
+ }
+
+ /** The location of the key store file for two-way authentication. */
+ public MongoDBSourceBuilder<T> sslKeyStore(String sslKeyStore) {
+ this.configFactory.sslKeyStore(sslKeyStore);
+ return this;
+ }
+
+ /** The password for the key store file. */
+ public MongoDBSourceBuilder<T> sslKeyStorePassword(String
sslKeyStorePassword) {
+ this.configFactory.sslKeyStorePassword(sslKeyStorePassword);
+ return this;
+ }
+
+ /** The type of key store file. */
+ public MongoDBSourceBuilder<T> sslKeyStoreType(String sslKeyStoreType) {
+ this.configFactory.sslKeyStoreType(sslKeyStoreType);
+ return this;
+ }
+
+ /** The location of the trust store file for the server certificate
verification. */
+ public MongoDBSourceBuilder<T> sslTrustStore(String sslTrustStore) {
+ this.configFactory.sslTrustStore(sslTrustStore);
+ return this;
+ }
+
+ /** The password for the trust store file. */
+ public MongoDBSourceBuilder<T> sslTrustStorePassword(String
sslTrustStorePassword) {
+ this.configFactory.sslTrustStorePassword(sslTrustStorePassword);
+ return this;
+ }
+
+ /** The type of trust store file. */
+ public MongoDBSourceBuilder<T> sslTrustStoreType(String sslTrustStoreType)
{
+ this.configFactory.sslTrustStoreType(sslTrustStoreType);
+ return this;
+ }
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
index b32dba452..df0a146a0 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java
@@ -59,6 +59,15 @@ public class MongoDBSourceConfig implements SourceConfig {
private final boolean assignUnboundedChunkFirst;
private final boolean releaseSnapshotMetadataEnabled;
+ private final boolean sslEnabled;
+ private final boolean sslInvalidHostnameAllowed;
+ @Nullable private final String sslKeyStore;
+ @Nullable private final String sslKeyStorePassword;
+ private final String sslKeyStoreType;
+ @Nullable private final String sslTrustStore;
+ @Nullable private final String sslTrustStorePassword;
+ private final String sslTrustStoreType;
+
MongoDBSourceConfig(
String scheme,
String hosts,
@@ -82,7 +91,15 @@ public class MongoDBSourceConfig implements SourceConfig {
boolean skipSnapshotBackfill,
boolean isScanNewlyAddedTableEnabled,
boolean assignUnboundedChunkFirst,
- boolean releaseSnapshotMetadataEnabled) {
+ boolean releaseSnapshotMetadataEnabled,
+ boolean sslEnabled,
+ boolean sslInvalidHostnameAllowed,
+ @Nullable String sslKeyStore,
+ @Nullable String sslKeyStorePassword,
+ String sslKeyStoreType,
+ @Nullable String sslTrustStore,
+ @Nullable String sslTrustStorePassword,
+ String sslTrustStoreType) {
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
this.username = username;
@@ -113,6 +130,14 @@ public class MongoDBSourceConfig implements SourceConfig {
+ "scan.newly-added-table.enabled cannot both be
enabled: releasing the "
+ "snapshot split metadata would drop the assigned
splits, finished offsets "
+ "and table schemas that newly-added-table scanning
needs.");
+ this.sslEnabled = sslEnabled;
+ this.sslInvalidHostnameAllowed = sslInvalidHostnameAllowed;
+ this.sslKeyStore = sslKeyStore;
+ this.sslKeyStorePassword = sslKeyStorePassword;
+ this.sslKeyStoreType = sslKeyStoreType;
+ this.sslTrustStore = sslTrustStore;
+ this.sslTrustStorePassword = sslTrustStorePassword;
+ this.sslTrustStoreType = sslTrustStoreType;
}
public String getScheme() {
@@ -217,6 +242,42 @@ public class MongoDBSourceConfig implements SourceConfig {
return assignUnboundedChunkFirst;
}
+ public boolean isSslEnabled() {
+ return sslEnabled;
+ }
+
+ public boolean isSslInvalidHostnameAllowed() {
+ return sslInvalidHostnameAllowed;
+ }
+
+ @Nullable
+ public String getSslKeyStore() {
+ return sslKeyStore;
+ }
+
+ @Nullable
+ public String getSslKeyStorePassword() {
+ return sslKeyStorePassword;
+ }
+
+ public String getSslKeyStoreType() {
+ return sslKeyStoreType;
+ }
+
+ @Nullable
+ public String getSslTrustStore() {
+ return sslTrustStore;
+ }
+
+ @Nullable
+ public String getSslTrustStorePassword() {
+ return sslTrustStorePassword;
+ }
+
+ public String getSslTrustStoreType() {
+ return sslTrustStoreType;
+ }
+
@Override
public boolean isReleaseSnapshotMetadataEnabled() {
return releaseSnapshotMetadataEnabled;
@@ -250,7 +311,15 @@ public class MongoDBSourceConfig implements SourceConfig {
&& Objects.equals(connectionString, that.connectionString)
&& Objects.equals(skipSnapshotBackfill,
that.skipSnapshotBackfill)
&& Objects.equals(isScanNewlyAddedTableEnabled,
that.isScanNewlyAddedTableEnabled)
- && releaseSnapshotMetadataEnabled ==
that.releaseSnapshotMetadataEnabled;
+ && releaseSnapshotMetadataEnabled ==
that.releaseSnapshotMetadataEnabled
+ && sslEnabled == that.sslEnabled
+ && sslInvalidHostnameAllowed == that.sslInvalidHostnameAllowed
+ && Objects.equals(sslKeyStore, that.sslKeyStore)
+ && Objects.equals(sslKeyStorePassword,
that.sslKeyStorePassword)
+ && Objects.equals(sslKeyStoreType, that.sslKeyStoreType)
+ && Objects.equals(sslTrustStore, that.sslTrustStore)
+ && Objects.equals(sslTrustStorePassword,
that.sslTrustStorePassword)
+ && Objects.equals(sslTrustStoreType, that.sslTrustStoreType);
}
@Override
@@ -275,6 +344,14 @@ public class MongoDBSourceConfig implements SourceConfig {
closeIdleReaders,
skipSnapshotBackfill,
isScanNewlyAddedTableEnabled,
- releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
+ sslEnabled,
+ sslInvalidHostnameAllowed,
+ sslKeyStore,
+ sslKeyStorePassword,
+ sslKeyStoreType,
+ sslTrustStore,
+ sslTrustStorePassword,
+ sslTrustStoreType);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
index 76782e822..3478ae5a5 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java
@@ -67,6 +67,17 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
protected boolean releaseSnapshotMetadataEnabled =
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED.defaultValue();
+ private boolean sslEnabled =
MongoDBSourceOptions.SSL_ENABLED.defaultValue();
+ private boolean sslInvalidHostnameAllowed =
+ MongoDBSourceOptions.SSL_INVALID_HOSTNAME_ALLOWED.defaultValue();
+ private String sslKeyStore =
MongoDBSourceOptions.SSL_KEYSTORE.defaultValue();
+ private String sslKeyStorePassword =
MongoDBSourceOptions.SSL_KEYSTORE_PASSWORD.defaultValue();
+ private String sslKeyStoreType =
MongoDBSourceOptions.SSL_KEYSTORE_TYPE.defaultValue();
+ private String sslTrustStore =
MongoDBSourceOptions.SSL_TRUSTSTORE.defaultValue();
+ private String sslTrustStorePassword =
+ MongoDBSourceOptions.SSL_TRUSTSTORE_PASSWORD.defaultValue();
+ private String sslTrustStoreType =
MongoDBSourceOptions.SSL_TRUSTSTORE_TYPE.defaultValue();
+
/** The protocol connected to MongoDB. For example mongodb or mongodb+srv.
*/
public MongoDBSourceConfigFactory scheme(String scheme) {
checkArgument(
@@ -293,6 +304,54 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
return this;
}
+ /** Whether the connector will use SSL to connect to MongoDB instances. */
+ public MongoDBSourceConfigFactory sslEnabled(boolean sslEnabled) {
+ this.sslEnabled = sslEnabled;
+ return this;
+ }
+
+ /** When SSL is enabled, controls whether strict hostname checking is
disabled. */
+ public MongoDBSourceConfigFactory sslInvalidHostnameAllowed(boolean
sslInvalidHostnameAllowed) {
+ this.sslInvalidHostnameAllowed = sslInvalidHostnameAllowed;
+ return this;
+ }
+
+ /** The location of the key store file for two-way authentication. */
+ public MongoDBSourceConfigFactory sslKeyStore(String sslKeyStore) {
+ this.sslKeyStore = sslKeyStore;
+ return this;
+ }
+
+ /** The password for the key store file. */
+ public MongoDBSourceConfigFactory sslKeyStorePassword(String
sslKeyStorePassword) {
+ this.sslKeyStorePassword = sslKeyStorePassword;
+ return this;
+ }
+
+ /** The type of key store file. */
+ public MongoDBSourceConfigFactory sslKeyStoreType(String sslKeyStoreType) {
+ this.sslKeyStoreType = sslKeyStoreType;
+ return this;
+ }
+
+ /** The location of the trust store file for the server certificate
verification. */
+ public MongoDBSourceConfigFactory sslTrustStore(String sslTrustStore) {
+ this.sslTrustStore = sslTrustStore;
+ return this;
+ }
+
+ /** The password for the trust store file. */
+ public MongoDBSourceConfigFactory sslTrustStorePassword(String
sslTrustStorePassword) {
+ this.sslTrustStorePassword = sslTrustStorePassword;
+ return this;
+ }
+
+ /** The type of trust store file. */
+ public MongoDBSourceConfigFactory sslTrustStoreType(String
sslTrustStoreType) {
+ this.sslTrustStoreType = sslTrustStoreType;
+ return this;
+ }
+
/** Creates a new {@link MongoDBSourceConfig} for the given subtask {@code
subtaskId}. */
@Override
public MongoDBSourceConfig create(int subtaskId) {
@@ -320,6 +379,14 @@ public class MongoDBSourceConfigFactory implements
Factory<MongoDBSourceConfig>
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
assignUnboundedChunkFirst,
- releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
+ sslEnabled,
+ sslInvalidHostnameAllowed,
+ sslKeyStore,
+ sslKeyStorePassword,
+ sslKeyStoreType,
+ sslTrustStore,
+ sslTrustStorePassword,
+ sslTrustStoreType);
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceOptions.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceOptions.java
index 3e4cf6096..402df011a 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceOptions.java
@@ -178,4 +178,68 @@ public class MongoDBSourceOptions {
.defaultValue(true)
.withDescription(
"MongoDB server normally times out idle cursors
after an inactivity period (10 minutes) to prevent excess memory use. Set this
option to true to prevent that.");
+
+ // SSL/TLS options
+
+ public static final ConfigOption<Boolean> SSL_ENABLED =
+ ConfigOptions.key("mongodb.ssl.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether the connector will use SSL to connect to
MongoDB instances.");
+
+ public static final ConfigOption<Boolean> SSL_INVALID_HOSTNAME_ALLOWED =
+ ConfigOptions.key("mongodb.ssl.invalid.hostname.allowed")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When SSL is enabled, this setting controls
whether strict hostname checking is disabled "
+ + "during the connection phase. If true,
the connection will not prevent man-in-the-middle attacks.");
+
+ public static final ConfigOption<String> SSL_KEYSTORE =
+ ConfigOptions.key("mongodb.ssl.keystore")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location of the key store file. "
+ + "This is optional and can be used for
two-way authentication between the client and the MongoDB server.");
+
+ public static final ConfigOption<String> SSL_KEYSTORE_PASSWORD =
+ ConfigOptions.key("mongodb.ssl.keystore.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The password for the key store file. "
+ + "This is optional and only needed if
'mongodb.ssl.keystore' is configured.");
+
+ public static final ConfigOption<String> SSL_KEYSTORE_TYPE =
+ ConfigOptions.key("mongodb.ssl.keystore.type")
+ .stringType()
+ .defaultValue("PKCS12")
+ .withDescription(
+ "The type of key store file. "
+ + "This is optional and only needed if
'mongodb.ssl.keystore' is configured. Defaults to PKCS12.");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE =
+ ConfigOptions.key("mongodb.ssl.truststore")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The location of the trust store file for the
server certificate verification.");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE_PASSWORD =
+ ConfigOptions.key("mongodb.ssl.truststore.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The password for the trust store file. "
+ + "Used to check the integrity of the
truststore, and unlock the truststore.");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE_TYPE =
+ ConfigOptions.key("mongodb.ssl.truststore.type")
+ .stringType()
+ .defaultValue("PKCS12")
+ .withDescription(
+ "The type of trust store file. "
+ + "This is optional and only needed if
'mongodb.ssl.truststore' is configured. Defaults to PKCS12.");
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/connection/MongoClientPool.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/connection/MongoClientPool.java
index 40fe9e6e8..4902f7857 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/connection/MongoClientPool.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/connection/MongoClientPool.java
@@ -18,13 +18,18 @@
package org.apache.flink.cdc.connectors.mongodb.source.connection;
import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
+import org.apache.flink.cdc.connectors.mongodb.source.utils.MongoDBSslUtils;
import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.net.ssl.SSLContext;
+
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -34,7 +39,7 @@ public class MongoClientPool {
private static final Logger LOG =
LoggerFactory.getLogger(MongoClientPool.class);
private static final MongoClientPool INSTANCE = new MongoClientPool();
- private final Map<String, MongoClient> pools = new ConcurrentHashMap<>();
+ private final Map<MongoDBSourceConfig, MongoClient> pools = new
ConcurrentHashMap<>();
private MongoClientPool() {}
@@ -43,15 +48,44 @@ public class MongoClientPool {
}
public MongoClient getOrCreateMongoClient(MongoDBSourceConfig
sourceConfig) {
+ if (!sourceConfig.isSslEnabled()
+ && (sourceConfig.getSslKeyStore() != null
+ || sourceConfig.getSslTrustStore() != null)) {
+ LOG.warn(
+ "{} or {} is configured, but {} is false. "
+ + "The keystore and truststore settings will be
ignored and the connection will be established without SSL.",
+ MongoDBSourceOptions.SSL_KEYSTORE.key(),
+ MongoDBSourceOptions.SSL_TRUSTSTORE.key(),
+ MongoDBSourceOptions.SSL_ENABLED.key());
+ }
+
return pools.computeIfAbsent(
- sourceConfig.getConnectionString(),
- rawConnectionString -> {
- ConnectionString connectionString = new
ConnectionString(rawConnectionString);
+ sourceConfig,
+ config -> {
+ ConnectionString connectionString =
+ new ConnectionString(config.getConnectionString());
LOG.info(
"Create and register mongo client {}@{}",
connectionString.getUsername(),
connectionString.getHosts());
- return MongoClients.create(connectionString);
+
+ MongoClientSettings.Builder settingsBuilder =
+
MongoClientSettings.builder().applyConnectionString(connectionString);
+
+ if (config.isSslEnabled()) {
+ SSLContext sslContext =
MongoDBSslUtils.createSSLContext(config);
+ settingsBuilder.applyToSslSettings(
+ builder -> {
+ builder.enabled(true);
+ builder.invalidHostNameAllowed(
+
config.isSslInvalidHostnameAllowed());
+ if (sslContext != null) {
+ builder.context(sslContext);
+ }
+ });
+ }
+
+ return MongoClients.create(settingsBuilder.build());
});
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtils.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtils.java
new file mode 100644
index 000000000..659e47b2e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtils.java
@@ -0,0 +1,121 @@
+/*
+ * 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.flink.cdc.connectors.mongodb.source.utils;
+
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
+
+import javax.annotation.Nullable;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+
+/** Utilities for building SSL context from MongoDB connector configuration. */
+public class MongoDBSslUtils {
+
+ private MongoDBSslUtils() {}
+
+ /**
+ * Creates a custom {@link SSLContext} from the given {@link
MongoDBSourceConfig}. If neither
+ * keystore nor truststore is configured, returns {@code null} which lets
the driver use the JVM
+ * default SSL context.
+ */
+ @Nullable
+ public static SSLContext createSSLContext(MongoDBSourceConfig config) {
+ if (!config.isSslEnabled()) {
+ return null;
+ }
+
+ String keyStorePath = config.getSslKeyStore();
+ String trustStorePath = config.getSslTrustStore();
+
+ if (keyStorePath == null && trustStorePath == null) {
+ return null;
+ }
+
+ try {
+ KeyManager[] keyManagers = null;
+ if (keyStorePath != null) {
+ char[] keyStorePassword =
+ config.getSslKeyStorePassword() != null
+ ? config.getSslKeyStorePassword().toCharArray()
+ : null;
+ KeyStore ks =
+ loadKeyStore(
+ config.getSslKeyStoreType(),
+ Paths.get(keyStorePath),
+ keyStorePassword);
+ KeyManagerFactory kmf =
+
KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ kmf.init(ks, keyStorePassword);
+ keyManagers = kmf.getKeyManagers();
+ }
+
+ TrustManager[] trustManagers = null;
+ if (trustStorePath != null) {
+ char[] trustStorePassword =
+ config.getSslTrustStorePassword() != null
+ ?
config.getSslTrustStorePassword().toCharArray()
+ : null;
+ KeyStore ts =
+ loadKeyStore(
+ config.getSslTrustStoreType(),
+ Paths.get(trustStorePath),
+ trustStorePassword);
+ TrustManagerFactory tmf =
+
TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(ts);
+ trustManagers = tmf.getTrustManagers();
+ }
+
+ SSLContext context = SSLContext.getInstance("TLS");
+ context.init(keyManagers, trustManagers, null);
+ return context;
+ } catch (NoSuchAlgorithmException | KeyStoreException |
UnrecoverableKeyException e) {
+ throw new RuntimeException("Unable to create KeyStore/TrustStore
manager factory", e);
+ } catch (KeyManagementException e) {
+ throw new RuntimeException("Unable to initialize SSL context", e);
+ }
+ }
+
+ private static KeyStore loadKeyStore(String type, Path path, @Nullable
char[] password) {
+ try (InputStream in = Files.newInputStream(path)) {
+ KeyStore ks = KeyStore.getInstance(type);
+ ks.load(in, password);
+ return ks;
+ } catch (IOException
+ | KeyStoreException
+ | NoSuchAlgorithmException
+ | CertificateException e) {
+ throw new RuntimeException("Unable to read key store file from '"
+ path + "'", e);
+ }
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
index d8aef1b5c..e7c2567c4 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSource.java
@@ -91,6 +91,15 @@ public class MongoDBTableSource implements ScanTableSource,
SupportsReadingMetad
private final boolean assignUnboundedChunkFirst;
private final boolean releaseSnapshotMetadataEnabled;
+ private final boolean sslEnabled;
+ private final boolean sslInvalidHostnameAllowed;
+ @Nullable private final String sslKeyStore;
+ @Nullable private final String sslKeyStorePassword;
+ private final String sslKeyStoreType;
+ @Nullable private final String sslTrustStore;
+ @Nullable private final String sslTrustStorePassword;
+ private final String sslTrustStoreType;
+
//
--------------------------------------------------------------------------------------------
// Mutable attributes
//
--------------------------------------------------------------------------------------------
@@ -129,7 +138,15 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
boolean skipSnapshotBackfill,
boolean scanNewlyAddedTableEnabled,
boolean assignUnboundedChunkFirst,
- boolean releaseSnapshotMetadataEnabled) {
+ boolean releaseSnapshotMetadataEnabled,
+ boolean sslEnabled,
+ boolean sslInvalidHostnameAllowed,
+ @Nullable String sslKeyStore,
+ @Nullable String sslKeyStorePassword,
+ String sslKeyStoreType,
+ @Nullable String sslTrustStore,
+ @Nullable String sslTrustStorePassword,
+ String sslTrustStoreType) {
this.physicalSchema = physicalSchema;
this.scheme = checkNotNull(scheme);
this.hosts = checkNotNull(hosts);
@@ -160,6 +177,14 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
this.releaseSnapshotMetadataEnabled = releaseSnapshotMetadataEnabled;
+ this.sslEnabled = sslEnabled;
+ this.sslInvalidHostnameAllowed = sslInvalidHostnameAllowed;
+ this.sslKeyStore = sslKeyStore;
+ this.sslKeyStorePassword = sslKeyStorePassword;
+ this.sslKeyStoreType = sslKeyStoreType;
+ this.sslTrustStore = sslTrustStore;
+ this.sslTrustStorePassword = sslTrustStorePassword;
+ this.sslTrustStoreType = sslTrustStoreType;
}
@Override
@@ -236,6 +261,16 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
Optional.ofNullable(splitMetaGroupSize).ifPresent(builder::splitMetaGroupSize);
Optional.ofNullable(splitSizeMB).ifPresent(builder::splitSizeMB);
Optional.ofNullable(samplesPerChunk).ifPresent(builder::samplesPerChunk);
+
+ builder.sslEnabled(sslEnabled)
+ .sslInvalidHostnameAllowed(sslInvalidHostnameAllowed)
+ .sslKeyStoreType(sslKeyStoreType)
+ .sslTrustStoreType(sslTrustStoreType);
+ Optional.ofNullable(sslKeyStore).ifPresent(builder::sslKeyStore);
+
Optional.ofNullable(sslKeyStorePassword).ifPresent(builder::sslKeyStorePassword);
+
Optional.ofNullable(sslTrustStore).ifPresent(builder::sslTrustStore);
+
Optional.ofNullable(sslTrustStorePassword).ifPresent(builder::sslTrustStorePassword);
+
return SourceProvider.of(builder.build());
} else {
org.apache.flink.cdc.connectors.mongodb.MongoDBSource.Builder<RowData> builder =
@@ -329,7 +364,15 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
assignUnboundedChunkFirst,
- releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
+ sslEnabled,
+ sslInvalidHostnameAllowed,
+ sslKeyStore,
+ sslKeyStorePassword,
+ sslKeyStoreType,
+ sslTrustStore,
+ sslTrustStorePassword,
+ sslTrustStoreType);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
@@ -374,7 +417,15 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
&& Objects.equals(scanNewlyAddedTableEnabled,
that.scanNewlyAddedTableEnabled)
&& Objects.equals(assignUnboundedChunkFirst,
that.assignUnboundedChunkFirst)
&& Objects.equals(
- releaseSnapshotMetadataEnabled,
that.releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
that.releaseSnapshotMetadataEnabled)
+ && sslEnabled == that.sslEnabled
+ && sslInvalidHostnameAllowed == that.sslInvalidHostnameAllowed
+ && Objects.equals(sslKeyStore, that.sslKeyStore)
+ && Objects.equals(sslKeyStorePassword,
that.sslKeyStorePassword)
+ && Objects.equals(sslKeyStoreType, that.sslKeyStoreType)
+ && Objects.equals(sslTrustStore, that.sslTrustStore)
+ && Objects.equals(sslTrustStorePassword,
that.sslTrustStorePassword)
+ && Objects.equals(sslTrustStoreType, that.sslTrustStoreType);
}
@Override
@@ -409,7 +460,15 @@ public class MongoDBTableSource implements
ScanTableSource, SupportsReadingMetad
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
assignUnboundedChunkFirst,
- releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
+ sslEnabled,
+ sslInvalidHostnameAllowed,
+ sslKeyStore,
+ sslKeyStorePassword,
+ sslKeyStoreType,
+ sslTrustStore,
+ sslTrustStorePassword,
+ sslTrustStoreType);
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
index c85bf9111..979cd1df1 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableSourceFactory.java
@@ -60,6 +60,14 @@ import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourc
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCAN_NO_CURSOR_TIMEOUT;
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCHEME;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_ENABLED;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_INVALID_HOSTNAME_ALLOWED;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE_PASSWORD;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE_TYPE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE_PASSWORD;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE_TYPE;
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.USERNAME;
import static
org.apache.flink.cdc.debezium.utils.ResolvedSchemaUtils.getPhysicalSchema;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -147,6 +155,15 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
checkArgument(physicalSchema.getPrimaryKey().isPresent(), "Primary key
must be present");
checkPrimaryKey(physicalSchema.getPrimaryKey().get(), "Primary key
must be _id field");
+ boolean sslEnabled = config.get(SSL_ENABLED);
+ boolean sslInvalidHostnameAllowed =
config.get(SSL_INVALID_HOSTNAME_ALLOWED);
+ String sslKeyStore = config.getOptional(SSL_KEYSTORE).orElse(null);
+ String sslKeyStorePassword =
config.getOptional(SSL_KEYSTORE_PASSWORD).orElse(null);
+ String sslKeyStoreType = config.get(SSL_KEYSTORE_TYPE);
+ String sslTrustStore = config.getOptional(SSL_TRUSTSTORE).orElse(null);
+ String sslTrustStorePassword =
config.getOptional(SSL_TRUSTSTORE_PASSWORD).orElse(null);
+ String sslTrustStoreType = config.get(SSL_TRUSTSTORE_TYPE);
+
OptionUtils.printOptions(IDENTIFIER, ((Configuration) config).toMap());
return new MongoDBTableSource(
@@ -177,7 +194,15 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
skipSnapshotBackfill,
scanNewlyAddedTableEnabled,
assignUnboundedChunkFirst,
- releaseSnapshotMetadataEnabled);
+ releaseSnapshotMetadataEnabled,
+ sslEnabled,
+ sslInvalidHostnameAllowed,
+ sslKeyStore,
+ sslKeyStorePassword,
+ sslKeyStoreType,
+ sslTrustStore,
+ sslTrustStorePassword,
+ sslTrustStoreType);
}
private void checkPrimaryKey(UniqueConstraint pk, String message) {
@@ -263,6 +288,14 @@ public class MongoDBTableSourceFactory implements
DynamicTableSourceFactory {
options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
options.add(SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED);
+ options.add(SSL_ENABLED);
+ options.add(SSL_INVALID_HOSTNAME_ALLOWED);
+ options.add(SSL_KEYSTORE);
+ options.add(SSL_KEYSTORE_PASSWORD);
+ options.add(SSL_KEYSTORE_TYPE);
+ options.add(SSL_TRUSTSTORE);
+ options.add(SSL_TRUSTSTORE_PASSWORD);
+ options.add(SSL_TRUSTSTORE_TYPE);
return options;
}
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSslConnectionITCase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSslConnectionITCase.java
new file mode 100644
index 000000000..f4837e5d6
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSslConnectionITCase.java
@@ -0,0 +1,410 @@
+/*
+ * 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.flink.cdc.connectors.mongodb.source;
+
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfigFactory;
+import org.apache.flink.cdc.connectors.mongodb.source.utils.MongoDBSslUtils;
+import org.apache.flink.cdc.connectors.mongodb.utils.MongoDBContainer;
+import org.apache.flink.cdc.connectors.mongodb.utils.MongoDBSslContainer;
+import org.apache.flink.cdc.connectors.utils.ExternalResourceProxy;
+import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.CloseableIterator;
+
+import com.mongodb.ConnectionString;
+import com.mongodb.MongoClientSettings;
+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.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import javax.net.ssl.SSLContext;
+
+import java.net.URL;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Integration tests verifying MongoDB with custom SSL context. */
+@Testcontainers
+class MongoDBSslConnectionITCase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MongoDBSslConnectionITCase.class);
+
+ private static final String MONGO_VERSION = getMongoVersion();
+
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ private static MongoDBSslContainer mongo;
+
+ private static String truststorePath;
+ private static String keystorePath;
+
+ @RegisterExtension
+ public final ExternalResourceProxy<MiniClusterWithClientResource>
miniClusterResource =
+ new ExternalResourceProxy<>(
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+
.setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+
.setRpcServiceSharing(RpcServiceSharing.DEDICATED)
+ .withHaLeadershipControl()
+ .build()));
+
+ @BeforeAll
+ static void startContainer() throws Exception {
+ truststorePath = resourcePath("ssl/client-truststore.p12");
+ keystorePath = resourcePath("ssl/client-keystore.p12");
+
+ mongo =
+ new MongoDBSslContainer("mongo:" + MONGO_VERSION)
+ .withLogConsumer(
+ frame -> LOG.info("[mongodb-tls] {}",
frame.getUtf8String()));
+ mongo.start();
+ }
+
+ @AfterAll
+ static void stopContainer() {
+ if (mongo != null) {
+ mongo.stop();
+ }
+ }
+
+ @Test
+ void truststoreOnlyConnectsAndReadWrites() {
+ MongoDBSourceConfig config =
+ configFactory()
+ .sslEnabled(true)
+ .sslTrustStore(truststorePath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .sslInvalidHostnameAllowed(true)
+ .create(0);
+
+ assertRoundTrip(config, "ssl_truststore_only");
+ }
+
+ @Test
+ void keystoreAndTruststoreConnectsAndReadWrites() {
+ MongoDBSourceConfig config =
+ configFactory()
+ .sslEnabled(true)
+ .sslKeyStore(keystorePath)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .sslTrustStore(truststorePath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .sslInvalidHostnameAllowed(true)
+ .create(0);
+
+ assertRoundTrip(config, "ssl_keystore_truststore");
+ }
+
+ @Test
+ void plainConnectionFailsAgainstTlsServer() {
+ String plainUri =
+ String.format(
+ "mongodb://%s:%d/",
+ mongo.getHost(),
mongo.getMappedPort(MongoDBSslContainer.MONGODB_PORT));
+ MongoClientSettings settings =
+ MongoClientSettings.builder()
+ .applyConnectionString(new ConnectionString(plainUri))
+ .build();
+
+ assertThatThrownBy(
+ () -> {
+ try (MongoClient client =
MongoClients.create(settings)) {
+ client.getDatabase("admin").runCommand(new
Document("ping", 1));
+ }
+ })
+ .isInstanceOf(Exception.class);
+ }
+
+ @Test
+ void jvmDefaultTruststoreIsUsedWhenNoStoresConfigured() throws Exception {
+ // Simulate the effect of -Djavax.net.ssl.trustStore=... at JVM
startup by installing
+ // a custom default SSLContext that trusts our test CA. The driver uses
+ // SSLContext.getDefault() when no keystore/truststore is configured
in the connector.
+ withDefaultSslContext(
+ buildSslContextFromTruststore(truststorePath,
"truststorepass"),
+ () -> {
+ MongoDBSourceConfig config =
+ configFactory()
+ .sslEnabled(true)
+ .sslInvalidHostnameAllowed(true)
+ .create(0);
+
+ // createSSLContext returns null when no
keystore/truststore is configured
+
assertThat(MongoDBSslUtils.createSSLContext(config)).isNull();
+
+ // Driver falls back to JVM default SSLContext, which now
trusts our CA
+ MongoClientSettings settings = buildSettings(config, null);
+ try (MongoClient client = MongoClients.create(settings)) {
+ Document result =
+ client.getDatabase("admin").runCommand(new
Document("ping", 1));
+ assertThat(result.getDouble("ok")).isEqualTo(1.0);
+ }
+ });
+ }
+
+ @Test
+ void connectorTruststoreOverridesJvmDefault() throws Exception {
+ // Install a default SSLContext that trusts nothing, simulating a
hostile JVM default.
+ // The connector's explicit truststore config must override it for the
connection to
+ // succeed.
+ withDefaultSslContext(
+ emptyTrustingSslContext(),
+ () -> {
+ MongoDBSourceConfig config =
+ configFactory()
+ .sslEnabled(true)
+ .sslTrustStore(truststorePath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .sslInvalidHostnameAllowed(true)
+ .create(0);
+
+
assertThat(MongoDBSslUtils.createSSLContext(config)).isNotNull();
+ assertCdcSourceReadsData(config,
"ssl_connector_overrides_jvm");
+ });
+ }
+
+ @Test
+ void connectorKeystoreTruststoreOverridesJvmDefault() throws Exception {
+ // Same as above but with both keystore and truststore configured in
the connector.
+ withDefaultSslContext(
+ emptyTrustingSslContext(),
+ () -> {
+ MongoDBSourceConfig config =
+ configFactory()
+ .sslEnabled(true)
+ .sslKeyStore(keystorePath)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .sslTrustStore(truststorePath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .sslInvalidHostnameAllowed(true)
+ .create(0);
+
+
assertThat(MongoDBSslUtils.createSSLContext(config)).isNotNull();
+ assertCdcSourceReadsData(config,
"ssl_connector_keystore_overrides_jvm");
+ });
+ }
+
+ /**
+ * Starts a real MongoDBSource CDC pipeline, writes a document, and
verifies that the source
+ * picks it up.
+ */
+ private void assertCdcSourceReadsData(MongoDBSourceConfig config, String
collectionName) {
+ String database = "ssl_cdc_test";
+ String fullName = database + "." + collectionName;
+
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ assertThat(ctx).isNotNull();
+
+ MongoClientSettings writeSettings = buildSettings(config, ctx);
+ try (MongoClient client = MongoClients.create(writeSettings)) {
+ MongoDatabase db = client.getDatabase(database);
+ db.getCollection(collectionName).insertOne(new Document("key",
"value"));
+ }
+
+ // Build and run a real CDC source using the same config
+ MongoDBSourceBuilder<String> builder =
+ MongoDBSource.<String>builder()
+ .hosts(
+ String.format(
+ "%s:%d",
+ mongo.getHost(),
+
mongo.getMappedPort(MongoDBSslContainer.MONGODB_PORT)))
+ .username(MongoDBContainer.FLINK_USER)
+ .password(MongoDBContainer.FLINK_USER_PASSWORD)
+ .databaseList(database)
+ .collectionList(fullName)
+ .sslEnabled(config.isSslEnabled())
+
.sslInvalidHostnameAllowed(config.isSslInvalidHostnameAllowed())
+ .deserializer(new JsonDebeziumDeserializationSchema());
+
+ if (config.getSslTrustStore() != null) {
+ builder.sslTrustStore(config.getSslTrustStore())
+ .sslTrustStorePassword(config.getSslTrustStorePassword())
+ .sslTrustStoreType(config.getSslTrustStoreType());
+ }
+ if (config.getSslKeyStore() != null) {
+ builder.sslKeyStore(config.getSslKeyStore())
+ .sslKeyStorePassword(config.getSslKeyStorePassword())
+ .sslKeyStoreType(config.getSslKeyStoreType());
+ }
+
+ MongoDBSource<String> source = builder.build();
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+ try (env;
+ CloseableIterator<String> iterator =
+ env.fromSource(
+ source,
+
org.apache.flink.api.common.eventtime.WatermarkStrategy
+ .noWatermarks(),
+ "MongoDB CDC SSL Source")
+ .executeAndCollect()) {
+ env.enableCheckpointing(1000);
+ env.setParallelism(1);
+
+ // We inserted exactly 1 document; snapshot phase will emit it.
Calling next()
+ // once avoids the blocking hasNext() loop.
+ String record = iterator.next();
+ LOG.info("CDC record: {}", record);
+ assertThat(record).contains(collectionName);
+ } catch (Exception e) {
+ LOG.debug("Error closing execution environment", e);
+ }
+ }
+
+ private void assertRoundTrip(MongoDBSourceConfig config, String
collectionName) {
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ assertThat(ctx).isNotNull();
+
+ MongoClientSettings settings = buildSettings(config, ctx);
+ try (MongoClient client = MongoClients.create(settings)) {
+ MongoDatabase db = client.getDatabase("ssl_integration_test");
+ MongoCollection<Document> col = db.getCollection(collectionName);
+
+ col.insertOne(new Document("hello", "world"));
+ List<Document> docs = col.find().into(new ArrayList<>());
+ assertThat(docs).hasSize(1);
+ assertThat(docs.get(0).getString("hello")).isEqualTo("world");
+ }
+ }
+
+ private MongoClientSettings buildSettings(MongoDBSourceConfig config,
SSLContext sslContext) {
+ String uri = mongo.getTlsConnectionString();
+ MongoClientSettings.Builder builder =
+ MongoClientSettings.builder()
+ .applyConnectionString(new ConnectionString(uri))
+ .applyToSslSettings(
+ ssl -> {
+ ssl.enabled(true);
+ ssl.invalidHostNameAllowed(
+
config.isSslInvalidHostnameAllowed());
+ if (sslContext != null) {
+ ssl.context(sslContext);
+ }
+ });
+ return builder.build();
+ }
+
+ private static MongoDBSourceConfigFactory configFactory() {
+ return new MongoDBSourceConfigFactory()
+ .hosts(
+ String.format(
+ "%s:%d",
+ mongo.getHost(),
+
mongo.getMappedPort(MongoDBSslContainer.MONGODB_PORT)));
+ }
+
+ private static String resourcePath(String classpathResource) throws
Exception {
+ URL url =
MongoDBSslConnectionITCase.class.getClassLoader().getResource(classpathResource);
+ Objects.requireNonNull(url, "Cannot locate test resource: " +
classpathResource);
+ return Paths.get(url.toURI()).toAbsolutePath().toString();
+ }
+
+ private static void withDefaultSslContext(SSLContext ctx, ThrowingRunnable
action)
+ throws Exception {
+ SSLContext previousDefault = SSLContext.getDefault();
+ try {
+ SSLContext.setDefault(ctx);
+ action.run();
+ } finally {
+ SSLContext.setDefault(previousDefault);
+ }
+ }
+
+ private static SSLContext buildSslContextFromTruststore(String path,
String password)
+ throws Exception {
+ java.security.KeyStore ts =
java.security.KeyStore.getInstance("PKCS12");
+ try (java.io.FileInputStream fis = new java.io.FileInputStream(path)) {
+ ts.load(fis, password.toCharArray());
+ }
+ javax.net.ssl.TrustManagerFactory tmf =
+ javax.net.ssl.TrustManagerFactory.getInstance(
+
javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm());
+ tmf.init(ts);
+ SSLContext ctx = SSLContext.getInstance("TLS");
+ ctx.init(null, tmf.getTrustManagers(), null);
+ return ctx;
+ }
+
+ private static SSLContext emptyTrustingSslContext() throws Exception {
+ SSLContext ctx = SSLContext.getInstance("TLS");
+ ctx.init(
+ null,
+ new javax.net.ssl.TrustManager[] {
+ new javax.net.ssl.X509TrustManager() {
+ public java.security.cert.X509Certificate[]
getAcceptedIssuers() {
+ return new java.security.cert.X509Certificate[0];
+ }
+
+ public void checkClientTrusted(
+ java.security.cert.X509Certificate[] c, String
a)
+ throws java.security.cert.CertificateException
{
+ throw new
java.security.cert.CertificateException("untrusted");
+ }
+
+ public void checkServerTrusted(
+ java.security.cert.X509Certificate[] c, String
a)
+ throws java.security.cert.CertificateException
{
+ throw new
java.security.cert.CertificateException("untrusted");
+ }
+ }
+ },
+ null);
+ return ctx;
+ }
+
+ @FunctionalInterface
+ private interface ThrowingRunnable {
+ void run() throws Exception;
+ }
+
+ private static String getMongoVersion() {
+ String v = System.getProperty("specifiedMongoVersion");
+ if (v == null || v.isBlank()) {
+ throw new IllegalArgumentException(
+ "No MongoDB version specified. Pass
-DspecifiedMongoVersion=<version>.");
+ }
+ return v;
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtilsTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtilsTest.java
new file mode 100644
index 000000000..8e327c9ec
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/utils/MongoDBSslUtilsTest.java
@@ -0,0 +1,294 @@
+/*
+ * 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.flink.cdc.connectors.mongodb.source.utils;
+
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfig;
+import
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceConfigFactory;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import javax.net.ssl.SSLContext;
+
+import java.net.URL;
+
+/** Unit tests for {@link MongoDBSslUtils}. */
+class MongoDBSslUtilsTest {
+
+ private static String keystoreP12Path;
+ private static String truststoreP12Path;
+ private static String keystoreJksPath;
+ private static String truststoreJksPath;
+
+ @BeforeAll
+ static void resolveResourcePaths() {
+ keystoreP12Path = resourcePath("ssl/client-keystore.p12");
+ truststoreP12Path = resourcePath("ssl/client-truststore.p12");
+ keystoreJksPath = resourcePath("ssl/client-keystore.jks");
+ truststoreJksPath = resourcePath("ssl/client-truststore.jks");
+ }
+
+ private static String resourcePath(String name) {
+ URL url = MongoDBSslUtilsTest.class.getClassLoader().getResource(name);
+ Assertions.assertThat(url).as("test resource '%s' not found",
name).isNotNull();
+ return url.getPath();
+ }
+
+ @Test
+ void sslDisabledReturnsNull() {
+ MongoDBSourceConfig config = factory().sslEnabled(false).create(0);
+
Assertions.assertThat(MongoDBSslUtils.createSSLContext(config)).isNull();
+ }
+
+ @Test
+ void sslDisabledWithKeystoreStillReturnsNull() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(false)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .create(0);
+
Assertions.assertThat(MongoDBSslUtils.createSSLContext(config)).isNull();
+ }
+
+ @Test
+ void sslDisabledWithTruststoreStillReturnsNull() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(false)
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+
Assertions.assertThat(MongoDBSslUtils.createSSLContext(config)).isNull();
+ }
+
+ @Test
+ void sslEnabledNoStoresReturnsNull() {
+ MongoDBSourceConfig config = factory().sslEnabled(true).create(0);
+
Assertions.assertThat(MongoDBSslUtils.createSSLContext(config)).isNull();
+ }
+
+ @Test
+ void truststoreOnlyPkcs12ProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void truststoreOnlyJksProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore(truststoreJksPath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("JKS")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void truststoreWithoutPasswordProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ }
+
+ @Test
+ void keystoreOnlyPkcs12ProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void keystoreOnlyJksProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreJksPath)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("JKS")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void keystoreAndTruststorePkcs12ProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void keystoreAndTruststoreJksProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreJksPath)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("JKS")
+ .sslTrustStore(truststoreJksPath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("JKS")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ Assertions.assertThat(ctx.getProtocol()).isEqualTo("TLS");
+ }
+
+ @Test
+ void keystoreP12AndTruststoreJksMixedTypesProducesContext() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .sslTrustStore(truststoreJksPath)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("JKS")
+ .create(0);
+ SSLContext ctx = MongoDBSslUtils.createSSLContext(config);
+ Assertions.assertThat(ctx).isNotNull();
+ }
+
+ @Test
+ void wrongKeystorePasswordThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("wrongpassword")
+ .sslKeyStoreType("PKCS12")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ @Test
+ void wrongTruststorePasswordThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStorePassword("wrongpassword")
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ @Test
+ void unknownKeystoreTypeThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore(keystoreP12Path)
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("BOGUSTYPE")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ @Test
+ void unknownTruststoreTypeThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore(truststoreP12Path)
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("BOGUSTYPE")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ @Test
+ void nonExistentKeystorePathThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslKeyStore("/nonexistent/path/keystore.p12")
+ .sslKeyStorePassword("keystorepass")
+ .sslKeyStoreType("PKCS12")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ @Test
+ void nonExistentTruststorePathThrows() {
+ MongoDBSourceConfig config =
+ factory()
+ .sslEnabled(true)
+ .sslTrustStore("/nonexistent/path/truststore.p12")
+ .sslTrustStorePassword("truststorepass")
+ .sslTrustStoreType("PKCS12")
+ .create(0);
+ Assertions.assertThatThrownBy(() ->
MongoDBSslUtils.createSSLContext(config))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Unable to read key store file");
+ }
+
+ private static MongoDBSourceConfigFactory factory() {
+ return new MongoDBSourceConfigFactory().hosts("localhost:27017");
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
index 12430d3a8..62a43d293 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/table/MongoDBTableFactoryTest.java
@@ -63,6 +63,14 @@ import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourc
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCAN_NO_CURSOR_TIMEOUT;
import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SCHEME;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_ENABLED;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_INVALID_HOSTNAME_ALLOWED;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE_PASSWORD;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_KEYSTORE_TYPE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE_PASSWORD;
+import static
org.apache.flink.cdc.connectors.mongodb.source.config.MongoDBSourceOptions.SSL_TRUSTSTORE_TYPE;
import static
org.apache.flink.cdc.connectors.utils.AssertUtils.assertProducedTypeOfSourceFunction;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -128,6 +136,17 @@ class MongoDBTableFactoryTest {
private static final boolean
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT =
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED.defaultValue();
+ private static final boolean SSL_ENABLED_DEFAULT =
SSL_ENABLED.defaultValue();
+ private static final boolean SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT =
+ SSL_INVALID_HOSTNAME_ALLOWED.defaultValue();
+ private static final String SSL_KEYSTORE_DEFAULT =
SSL_KEYSTORE.defaultValue();
+ private static final String SSL_KEYSTORE_PASSWORD_DEFAULT =
+ SSL_KEYSTORE_PASSWORD.defaultValue();
+ private static final String SSL_KEYSTORE_TYPE_DEFAULT =
SSL_KEYSTORE_TYPE.defaultValue();
+ private static final String SSL_TRUSTSTORE_DEFAULT =
SSL_TRUSTSTORE.defaultValue();
+ private static final String SSL_TRUSTSTORE_PASSWORD_DEFAULT =
+ SSL_TRUSTSTORE_PASSWORD.defaultValue();
+ private static final String SSL_TRUSTSTORE_TYPE_DEFAULT =
SSL_TRUSTSTORE_TYPE.defaultValue();
@Test
void testCommonProperties() {
@@ -164,7 +183,15 @@ class MongoDBTableFactoryTest {
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
-
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT,
+ SSL_ENABLED_DEFAULT,
+ SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT,
+ SSL_KEYSTORE_DEFAULT,
+ SSL_KEYSTORE_PASSWORD_DEFAULT,
+ SSL_KEYSTORE_TYPE_DEFAULT,
+ SSL_TRUSTSTORE_DEFAULT,
+ SSL_TRUSTSTORE_PASSWORD_DEFAULT,
+ SSL_TRUSTSTORE_TYPE_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
@@ -221,7 +248,15 @@ class MongoDBTableFactoryTest {
true,
true,
true,
-
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT,
+ SSL_ENABLED_DEFAULT,
+ SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT,
+ SSL_KEYSTORE_DEFAULT,
+ SSL_KEYSTORE_PASSWORD_DEFAULT,
+ SSL_KEYSTORE_TYPE_DEFAULT,
+ SSL_TRUSTSTORE_DEFAULT,
+ SSL_TRUSTSTORE_PASSWORD_DEFAULT,
+ SSL_TRUSTSTORE_TYPE_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
@@ -263,7 +298,15 @@ class MongoDBTableFactoryTest {
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
- true);
+ true,
+ SSL_ENABLED_DEFAULT,
+ SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT,
+ SSL_KEYSTORE_DEFAULT,
+ SSL_KEYSTORE_PASSWORD_DEFAULT,
+ SSL_KEYSTORE_TYPE_DEFAULT,
+ SSL_TRUSTSTORE_DEFAULT,
+ SSL_TRUSTSTORE_PASSWORD_DEFAULT,
+ SSL_TRUSTSTORE_TYPE_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
// equals has to carry the field, or a source built without the option
would compare equal
@@ -313,7 +356,15 @@ class MongoDBTableFactoryTest {
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
-
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT,
+ SSL_ENABLED_DEFAULT,
+ SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT,
+ SSL_KEYSTORE_DEFAULT,
+ SSL_KEYSTORE_PASSWORD_DEFAULT,
+ SSL_KEYSTORE_TYPE_DEFAULT,
+ SSL_TRUSTSTORE_DEFAULT,
+ SSL_TRUSTSTORE_PASSWORD_DEFAULT,
+ SSL_TRUSTSTORE_TYPE_DEFAULT);
expectedSource.producedDataType =
SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name",
"row_kind");
@@ -425,7 +476,15 @@ class MongoDBTableFactoryTest {
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP_DEFAULT,
SCAN_NEWLY_ADDED_TABLE_ENABLED_DEFAULT,
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED.defaultValue(),
-
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT);
+
SCAN_INCREMENTAL_SNAPSHOT_METADATA_RELEASE_ENABLED_DEFAULT,
+ SSL_ENABLED_DEFAULT,
+ SSL_INVALID_HOSTNAME_ALLOWED_DEFAULT,
+ SSL_KEYSTORE_DEFAULT,
+ SSL_KEYSTORE_PASSWORD_DEFAULT,
+ SSL_KEYSTORE_TYPE_DEFAULT,
+ SSL_TRUSTSTORE_DEFAULT,
+ SSL_TRUSTSTORE_PASSWORD_DEFAULT,
+ SSL_TRUSTSTORE_TYPE_DEFAULT);
Assertions.assertThat(actualSource).isEqualTo(expectedSource);
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/utils/MongoDBSslContainer.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/utils/MongoDBSslContainer.java
new file mode 100644
index 000000000..f7e5b18ea
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/utils/MongoDBSslContainer.java
@@ -0,0 +1,179 @@
+/*
+ * 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.flink.cdc.connectors.mongodb.utils;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.apache.commons.lang3.StringUtils;
+import org.assertj.core.api.Assertions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.OutputFrame;
+import org.testcontainers.utility.MountableFile;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.function.Consumer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * A MongoDB container configured to require TLS connections. Certificates are
taken from the
+ * test-resource ssl/ directory: server.pem (private key + cert) and ca.crt
(CA certificate).
+ *
+ * <p>The container mounts the server-side PEM and CA certificate into
/etc/ssl/ and starts mongod
+ * with --tlsMode requireTLS, so every client must present a valid TLS
handshake.
+ */
+public class MongoDBSslContainer extends
org.testcontainers.containers.MongoDBContainer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MongoDBSslContainer.class);
+
+ private static final Pattern COMMENT_PATTERN =
Pattern.compile("^(.*)//.*$");
+
+ public static final int MONGODB_PORT = 27017;
+
+ private static final String CONTAINER_SSL_DIR = "/etc/ssl/mongodb/";
+ private static final String CONTAINER_SERVER_PEM = CONTAINER_SSL_DIR +
"server.pem";
+ private static final String CONTAINER_CA_CRT = CONTAINER_SSL_DIR +
"ca.crt";
+
+ public MongoDBSslContainer(String imageName) {
+ super(imageName);
+ withCommand(
+ "--replSet",
+ "docker-rs",
+ "--tlsMode",
+ "requireTLS",
+ "--tlsCertificateKeyFile",
+ CONTAINER_SERVER_PEM,
+ "--tlsCAFile",
+ CONTAINER_CA_CRT,
+ // Allow clients that do not present a certificate
(truststore-only clients).
+ // Clients that do present one must still have it signed by
the CA.
+ "--tlsAllowConnectionsWithoutCertificates");
+ withCopyFileToContainer(
+ MountableFile.forClasspathResource("ssl/server.pem", 0444),
CONTAINER_SERVER_PEM);
+ withCopyFileToContainer(
+ MountableFile.forClasspathResource("ssl/ca.crt", 0444),
CONTAINER_CA_CRT);
+ }
+
+ @Override
+ protected void containerIsStarted(InspectContainerResponse containerInfo,
boolean reused) {
+ // Re-implement replica set init: the parent calls rs.initiate() via
mongosh without TLS
+ // flags, which would fail against a requireTLS server. We override it
to pass --tls.
+ initReplicaSetWithTls(reused);
+ createFlinkUser();
+ }
+
+ private void initReplicaSetWithTls(boolean reused) {
+ if (reused) {
+ return;
+ }
+ try {
+ ExecResult result =
+ execInContainer(
+ "mongosh",
+ "--tls",
+ "--tlsCertificateKeyFile",
+ CONTAINER_SERVER_PEM,
+ "--tlsCAFile",
+ CONTAINER_CA_CRT,
+ "--tlsAllowInvalidHostnames",
+ "--eval",
+ "rs.initiate()");
+ LOG.info(result.getStdout());
+ if (result.getExitCode() != 0) {
+ throw new IllegalStateException("rs.initiate() failed: " +
result.getStderr());
+ }
+ // Wait until the node becomes primary
+ for (int i = 0; i < 30; i++) {
+ ExecResult status =
+ execInContainer(
+ "mongosh",
+ "--tls",
+ "--tlsCertificateKeyFile",
+ CONTAINER_SERVER_PEM,
+ "--tlsCAFile",
+ CONTAINER_CA_CRT,
+ "--tlsAllowInvalidHostnames",
+ "--eval",
+ "db.isMaster().ismaster");
+ if (status.getStdout().contains("true")) {
+ return;
+ }
+ Thread.sleep(1000);
+ }
+ throw new IllegalStateException("Timed out waiting for replica set
primary");
+ } catch (InterruptedException | IOException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Failed to initialise replica set", e);
+ }
+ }
+
+ private void createFlinkUser() {
+ final String setupFilePath = "docker/mongodb/setup.js";
+ final URL setupFile =
MongoDBSslContainer.class.getClassLoader().getResource(setupFilePath);
+ Assertions.assertThat(setupFile)
+ .withFailMessage("Cannot locate " + setupFilePath)
+ .isNotNull();
+ try {
+ String createUserCommand =
+ Files.readAllLines(Paths.get(setupFile.toURI())).stream()
+ .filter(x -> StringUtils.isNotBlank(x) &&
!x.trim().startsWith("//"))
+ .map(
+ x -> {
+ final Matcher m =
COMMENT_PATTERN.matcher(x);
+ return m.matches() ? m.group(1) : x;
+ })
+ .collect(Collectors.joining(" "));
+ ExecResult execResult =
+ execInContainer(
+ "mongosh",
+ "--tls",
+ "--tlsCertificateKeyFile",
+ CONTAINER_SERVER_PEM,
+ "--tlsCAFile",
+ CONTAINER_CA_CRT,
+ "--tlsAllowInvalidHostnames",
+ "--eval",
+ "use admin",
+ "--eval",
+ createUserCommand,
+ "--eval",
+ "console.log('Flink test user created.\\n');");
+ LOG.info(execResult.getStdout());
+ if (execResult.getExitCode() != 0) {
+ throw new IllegalStateException(
+ "Failed to create Flink user: " +
execResult.getStderr());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public MongoDBSslContainer withLogConsumer(Consumer<OutputFrame> consumer)
{
+ return (MongoDBSslContainer) super.withLogConsumer(consumer);
+ }
+
+ /** Returns a connection URI with {@code tls=true} appended. */
+ public String getTlsConnectionString() {
+ return String.format("mongodb://%s:%d/?tls=true", getHost(),
getMappedPort(MONGODB_PORT));
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.crt
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.crt
new file mode 100644
index 000000000..a6b048cf1
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.crt
@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDUTCCAjmgAwIBAgIUfBOq6ZRTRVPpsMTzRDUz/hxhSCIwDQYJKoZIhvcNAQEL
+BQAwODEQMA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3Qx
+CzAJBgNVBAYTAkNOMB4XDTI2MDUxMTA3NDUzOVoXDTM2MDUwODA3NDUzOVowODEQ
+MA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3QxCzAJBgNV
+BAYTAkNOMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAs2sxiJ1oMHJz
+1b6RzHOplJ97NHiLmq0+x9bsSRJgeEQB/dOZ+t8N9xcnTIdcESGGRkWUPar1HRQm
+CmAgOKlSLUDeU/z28ageZ71bW/HYlDpfkTnpaNyFdSp985qwF+7Z8pxwIEt0v/I1
+/z7f+L9C6YrR86hNf0nRPOsRqtYx5X2L+BJQGvgb1ci9F6pqCr3vgfjDboIWId95
+p60aflYS/KxQGAieCbXSPT2aqAh9Xj6Bfhp/nZ44gbSd9k3LugSAmcd+6CODSA5K
+N01TwGhDAKoRIoKMXGUsfL1Tj1wYD9CGFnUzBunbE8UhzxaZfk+RnZlJf7V+PZM6
+KGoahmnNEQIDAQABo1MwUTAdBgNVHQ4EFgQUQKZcDIEbs1XT3/Q+gv2uGUjV+kgw
+HwYDVR0jBBgwFoAUQKZcDIEbs1XT3/Q+gv2uGUjV+kgwDwYDVR0TAQH/BAUwAwEB
+/zANBgkqhkiG9w0BAQsFAAOCAQEAKqVwBU83qEdn2kRKd2Wp8e06AiFCYb5RH3YM
+7bonRUop2QrVYAAfDwzPaG8Qqf9RDrLrItnUPYNnl2F6LQ8gODrjdnJDU75LGNVN
+9eAxsNbbYnnfayOq5XszO+Akcb4Z4Jcc+QKUz1wJpI6USLSeG3eIVXY8Ef7RIFzQ
+z7yR9WWqSrJWkN/WgIAMsoyQ8WZeiA0ONrzFZbIcExEzzM9Oap9jKFEZM3ZltS61
+O5OPe8BC0fnOY9vJNxdEwzl6hjxzQVuw/JihlzKoaayGHVgkOeazZATwTVGvD6Cn
+/tJUgOVaIazzcXYcdGSSQof4GirxHwEAGB0Y5hmsTBgHf6j64A==
+-----END CERTIFICATE-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.key
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.key
new file mode 100644
index 000000000..1e2ffcb54
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCzazGInWgwcnPV
+vpHMc6mUn3s0eIuarT7H1uxJEmB4RAH905n63w33FydMh1wRIYZGRZQ9qvUdFCYK
+YCA4qVItQN5T/PbxqB5nvVtb8diUOl+ROelo3IV1Kn3zmrAX7tnynHAgS3S/8jX/
+Pt/4v0LpitHzqE1/SdE86xGq1jHlfYv4ElAa+BvVyL0XqmoKve+B+MNughYh33mn
+rRp+VhL8rFAYCJ4JtdI9PZqoCH1ePoF+Gn+dnjiBtJ32Tcu6BICZx37oI4NIDko3
+TVPAaEMAqhEigoxcZSx8vVOPXBgP0IYWdTMG6dsTxSHPFpl+T5GdmUl/tX49kzoo
+ahqGac0RAgMBAAECggEAEHSJsSM+pbLZjNuoOOgiwCQgWJ+gWNSq0rPo1+AnmOAj
+4/O+8OikOeqVtXR0rTLgmHFT4elcBwliCRw9LZ/Z5EH0Ih/Wm4ifwUP3V7zGmvxU
+54I4NgsuR6CuDI7ot56LFAU6WLvoPp0oS/tS71gbUEVlxsMY2Hkcskmc794qlVws
+X24dDe6qXph+0ayUh8Cb+8Tf6auiqZfUyBKkcnesdKMFxFg1yWFiuz+a2oSz6/yD
+k7Nz6GIJRUArdyORKLGz4dH1EqoJ2ni5P9DcUTpacsA9b81I7aONFU5apUYnm+ZU
+Pb9T663jI4rXf+Su4H6IkEdPypW7c5vmnuHKUXxKQQKBgQDdh1A/1oO2e1477jBU
+aUmlQe7+YSTQ8KihHWZxXOMGR56ypM3QoevyUyipcuvvG5UvlU26lrfiVOx9ximF
+XjOd6M9ksP1no9gPFqxDIsdCIpEyTKx53C1lBMCkBfJZUtbX2MFTMmce93fBkiXs
++fM0VV3XfIZ5nhwDGUdiKWRoKwKBgQDPVmqIXza2G9PJMLbsjxAvJWzDwFKF71j8
+43Ul2LLs1gMtYLB9YVmZpAUCRFl/sX4uAp94MysX2iaXIaUrtsgsI26b8wqLNUCF
+fyXi3omQUtorKBAUFHiD2sPmWlvLXwW2svzyNwR9IO5pGBO1J8sOSLEbYwiZdrgy
+9ydTXlRlswKBgCioOPY+bJYFNlpRcNVNLZMC8tYFBjjCsnA/cRnFLnVsl5e533w2
++5LR6FiRGvQbX9oH0Og52S3U4EKFKAGuNFL683VLdnaa/4BlbOA1ddSRe4x7vfvv
+iPVNVf3EiaITLhMXWQPn8D0itfv00rkXdqdYM4zk8dlIR2PaM9u0NRVZAoGAcdxN
+CLLxArTpazUbX3F5O8rbnexB5Vd5kPcAYFZOfA1TFVph0ZjicaNHNOWAGMXKO04p
+cQL4iVK9moVZhVYNO2AB4BZqFeSamDUpAxueg+lb8hNAMSbEv/QlXJ0Gh/qY0qoc
+io94IZcyFfio6fT+Z8u3pwiNsmkl4qcjq37D7k0CgYBdZ8ZpZjAXZvC9JSQpvcY3
+NSgJsKTdYtjI9tchXdWDYQy0ley0pu3Rc5cgjICGaFlbtC4L6DFPQOTGTKcSyAG2
+s6/AULbtr9Ew+EQXNMtblYMSpF9C90Owdq7qHy3chZ5kTY2UW5dBf2/4oNnRmKq4
+ADE9xXqchxZ7aWgjuA2MZQ==
+-----END PRIVATE KEY-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.srl
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.srl
new file mode 100644
index 000000000..8827254fa
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/ca.srl
@@ -0,0 +1 @@
+6836288FF8790D3E2728A0BE6975CCA205B6DF11
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.jks
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.jks
new file mode 100644
index 000000000..0e96d5fd9
Binary files /dev/null and
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.jks
differ
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.p12
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.p12
new file mode 100644
index 000000000..9af501bcb
Binary files /dev/null and
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-keystore.p12
differ
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.jks
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.jks
new file mode 100644
index 000000000..6eb4ca061
Binary files /dev/null and
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.jks
differ
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.p12
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.p12
new file mode 100644
index 000000000..5e121e892
Binary files /dev/null and
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client-truststore.p12
differ
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.crt
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.crt
new file mode 100644
index 000000000..f2bd6f182
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.crt
@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDRDCCAiygAwIBAgIUaDYoj/h5DT4nKKC+aXXMogW23xEwDQYJKoZIhvcNAQEL
+BQAwODEQMA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3Qx
+CzAJBgNVBAYTAkNOMB4XDTI2MDUxMTA3NDU0MFoXDTM2MDUwODA3NDU0MFowPDEU
+MBIGA1UEAwwLVGVzdCBDbGllbnQxFzAVBgNVBAoMDkZsaW5rIENEQyBUZXN0MQsw
+CQYDVQQGEwJDTjCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAJdonYSK
+eHk/TDjwqBp4MMG1PipiemxUNIeQP7UFmBs8Ehr2F9Do8WFSl61PfkrZpJIjjVIm
+kw8nW47i49E11pjIKOfe6zpEqT1cYt6gOreQRujmK7C1H7RAHbwdRPQcTvLa6m5T
+p79waeDglbMHvDMrw7uJLq1qEFSz+g5FEoBaIcxwap3m4FgapcmlMrdKxZKyQBc6
+Kwho3+qhcbybWLYi2r3djduKx943XG7YBASjR3rVn94IeCF40qBsq4z0FCaWe4lx
+xasEXEOE3JSUXq1qKHFAEywmW4W427DTeNWeF0MZwNT8amEOFXN18gglY94piw+d
+n7US1h6qVCK4S80CAwEAAaNCMEAwHQYDVR0OBBYEFGhHtur41a1Ls20y0f2uavbb
+Aey3MB8GA1UdIwQYMBaAFECmXAyBG7NV09/0PoL9rhlI1fpIMA0GCSqGSIb3DQEB
+CwUAA4IBAQAihv58yUnc2P5tXevUB4g8LoC4ZH+nBX9Nl7ZWJX6a7fkA4kH5O/lL
+DVlS0Xs1NZ6MbYoSQTZnpJ1N/Lh/76ens2TxExznmwWy7+JJgpjPivpbbi5wIEwJ
+Vkn8RRlrCobRBSRacA2Jn11ZPEC7EqcNoF7t3SR1hPDT4EJngbggavhhW+F2mPY9
+B5ZROfFVEjSTlfXt6g/FVZ/HL7CC7vc82R5WCYvGkIps6bR+PHUqP4yNXTW6L/Ca
+qeEpJNjJ1SwmpVl++MTtSj1OPOOLzXXzmYHylEDAzNPQlnSIeh7AJsrUfeLwrUFO
+07EbLxeRUYa4VLfWDO/oL1deaMgyyr5M
+-----END CERTIFICATE-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.csr
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.csr
new file mode 100644
index 000000000..a129b0f92
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.csr
@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE REQUEST-----
+MIICgTCCAWkCAQAwPDEUMBIGA1UEAwwLVGVzdCBDbGllbnQxFzAVBgNVBAoMDkZs
+aW5rIENEQyBUZXN0MQswCQYDVQQGEwJDTjCCASIwDQYJKoZIhvcNAQEBBQADggEP
+ADCCAQoCggEBAJdonYSKeHk/TDjwqBp4MMG1PipiemxUNIeQP7UFmBs8Ehr2F9Do
+8WFSl61PfkrZpJIjjVImkw8nW47i49E11pjIKOfe6zpEqT1cYt6gOreQRujmK7C1
+H7RAHbwdRPQcTvLa6m5Tp79waeDglbMHvDMrw7uJLq1qEFSz+g5FEoBaIcxwap3m
+4FgapcmlMrdKxZKyQBc6Kwho3+qhcbybWLYi2r3djduKx943XG7YBASjR3rVn94I
+eCF40qBsq4z0FCaWe4lxxasEXEOE3JSUXq1qKHFAEywmW4W427DTeNWeF0MZwNT8
+amEOFXN18gglY94piw+dn7US1h6qVCK4S80CAwEAAaAAMA0GCSqGSIb3DQEBCwUA
+A4IBAQCSRx59xAxC9ZwzVz+uULn+lqA1/OezSRPzIxWtyQZMENSmjIzp06c3jwzl
+e8k72yXXXwvcU8RslnDEQRijqc+0ruWU5FjxqgVHfrZ8Q4YuyDrJu+pm8m9Vvnx3
+yAGgp0hHgnbKZuhFemOCEKvjYlWGS/d1Iygsy/gSB2y9lZuYyQ8WjXTnIrLlgrYJ
+/MZBVyt5oxHl/VMsFfGd2hO7ygDvTWXD7azzJR9jUOek1Cm5btCveVxCUXeD2iMu
+ZRojt79DAwOUahhJmNZLcEcaxA4gIddVQqBqiOWUgIn3HlfmiROpwsvoMQeWUlgL
+KlpFKPfe3u1A0oSXW/9y+HW/tLne
+-----END CERTIFICATE REQUEST-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.key
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.key
new file mode 100644
index 000000000..72e29e60e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCXaJ2Einh5P0w4
+8KgaeDDBtT4qYnpsVDSHkD+1BZgbPBIa9hfQ6PFhUpetT35K2aSSI41SJpMPJ1uO
+4uPRNdaYyCjn3us6RKk9XGLeoDq3kEbo5iuwtR+0QB28HUT0HE7y2upuU6e/cGng
+4JWzB7wzK8O7iS6tahBUs/oORRKAWiHMcGqd5uBYGqXJpTK3SsWSskAXOisIaN/q
+oXG8m1i2Itq93Y3bisfeN1xu2AQEo0d61Z/eCHgheNKgbKuM9BQmlnuJccWrBFxD
+hNyUlF6taihxQBMsJluFuNuw03jVnhdDGcDU/GphDhVzdfIIJWPeKYsPnZ+1EtYe
+qlQiuEvNAgMBAAECggEAAlCIIJVvb1GDXux3HPddWI6DC9uMmCraXw5FKNOiXyo2
+LE+B1cDEGIv4/HOhEFAkRRqlwHFGPdsXMusg2mIlSTYEkBFQBjMmAbrhWmGevCuW
+0lywalliCcZnQYUgsVy/+ozwkcnOw9DMCKvjlC9BAPbr5HFk5SPSIyVZh7h7UTqw
+mHX6/JfBpxhqePsntuA7H5Q3+uBqJqds2/MfKXX5FQ5FXyt4DStUxLUi2+vBWTsD
+eKQ2T9R+iglILWXcHVXDTIO5+j71gK4qlw/mk5sJb3+IqNOGh6ZNlzcKo9SIguPb
+0ZHuDiHsIKFtsQL/JKTcHTOxVZIypLe9ff1VkhjjkQKBgQDHIY1OTONPWhA+1cmc
+EGySVVFF7rUS5YuRavm+xcEAsWt2yB/+PCSGmgAHfP/PYQ/BMGLML9+UDhndrfyy
+GK3g7WrdOo18JXN1HfGx7zq70d4yv5FG+iJNFkFDyLQG69YPYnYhssl9kekOjdl9
+lS+RHvYsvrrJRRXe08iRwGy6UQKBgQDCphVw55yiPhwsEPnpFDQtXmG//rnaX763
+6cQ34wldDsTd2mXAzd76UCTKDXYvi9rV1whKzsu1JMVf6LX8JOpeZJFMP9FdUyRV
+1kIYKx+qCB9M3um9aIVN9s/HftS1hq9wo7JqyOm+PZELsjlV8RxI4fy2YhjuxsS3
+4L+sijlevQKBgDgxcazM6cB0y+zPa6tMvza0ONEhQZvphu8sXvnE2GgXqRnF87FM
+y4noX2W9knq1CzsbKmYwhElFiWsTNCIV8MtudB+l7FNCRkiggJoTxmFGIgv0H00G
+PTO/uf8k3hit6KRCxs9Poqh6yme/A8b6mIe9Uv7he9hcVIWjAhCPWp9hAoGASv84
+9aaXyD9MFYJ49+Ar9msOGK6pp0UKs8vjKXFVs0A0hBJY8D+iMSnd8ufa6JjnOUs/
+y6r//Axt6/U64SrMDpCLFyjjnySgtF34W12KNrYocTLWNn9mFVMKXaA1jreVMz4v
+SUueHAmWsD38L2vV7RL9jrwfRSjatMpor68WSbUCgYEAoH16QdkrFU6jvP9rcd62
+UjtzUvg3Lpk/727qiBbLOnqGUMT9dNelC7tHyEkv+EgBtIXi02WVIPXdYgW/nOqQ
+zBtkIdOF+rpVgwfmN6l6uASlp3nJNvLPm7i1kCr4XiXFvCkMDg1RRJul8wfC8+Uz
+wDMtidTTy+HHcN8z+RtAYTI=
+-----END PRIVATE KEY-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.pem
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.pem
new file mode 100644
index 000000000..1696969f5
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/client.pem
@@ -0,0 +1,48 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCXaJ2Einh5P0w4
+8KgaeDDBtT4qYnpsVDSHkD+1BZgbPBIa9hfQ6PFhUpetT35K2aSSI41SJpMPJ1uO
+4uPRNdaYyCjn3us6RKk9XGLeoDq3kEbo5iuwtR+0QB28HUT0HE7y2upuU6e/cGng
+4JWzB7wzK8O7iS6tahBUs/oORRKAWiHMcGqd5uBYGqXJpTK3SsWSskAXOisIaN/q
+oXG8m1i2Itq93Y3bisfeN1xu2AQEo0d61Z/eCHgheNKgbKuM9BQmlnuJccWrBFxD
+hNyUlF6taihxQBMsJluFuNuw03jVnhdDGcDU/GphDhVzdfIIJWPeKYsPnZ+1EtYe
+qlQiuEvNAgMBAAECggEAAlCIIJVvb1GDXux3HPddWI6DC9uMmCraXw5FKNOiXyo2
+LE+B1cDEGIv4/HOhEFAkRRqlwHFGPdsXMusg2mIlSTYEkBFQBjMmAbrhWmGevCuW
+0lywalliCcZnQYUgsVy/+ozwkcnOw9DMCKvjlC9BAPbr5HFk5SPSIyVZh7h7UTqw
+mHX6/JfBpxhqePsntuA7H5Q3+uBqJqds2/MfKXX5FQ5FXyt4DStUxLUi2+vBWTsD
+eKQ2T9R+iglILWXcHVXDTIO5+j71gK4qlw/mk5sJb3+IqNOGh6ZNlzcKo9SIguPb
+0ZHuDiHsIKFtsQL/JKTcHTOxVZIypLe9ff1VkhjjkQKBgQDHIY1OTONPWhA+1cmc
+EGySVVFF7rUS5YuRavm+xcEAsWt2yB/+PCSGmgAHfP/PYQ/BMGLML9+UDhndrfyy
+GK3g7WrdOo18JXN1HfGx7zq70d4yv5FG+iJNFkFDyLQG69YPYnYhssl9kekOjdl9
+lS+RHvYsvrrJRRXe08iRwGy6UQKBgQDCphVw55yiPhwsEPnpFDQtXmG//rnaX763
+6cQ34wldDsTd2mXAzd76UCTKDXYvi9rV1whKzsu1JMVf6LX8JOpeZJFMP9FdUyRV
+1kIYKx+qCB9M3um9aIVN9s/HftS1hq9wo7JqyOm+PZELsjlV8RxI4fy2YhjuxsS3
+4L+sijlevQKBgDgxcazM6cB0y+zPa6tMvza0ONEhQZvphu8sXvnE2GgXqRnF87FM
+y4noX2W9knq1CzsbKmYwhElFiWsTNCIV8MtudB+l7FNCRkiggJoTxmFGIgv0H00G
+PTO/uf8k3hit6KRCxs9Poqh6yme/A8b6mIe9Uv7he9hcVIWjAhCPWp9hAoGASv84
+9aaXyD9MFYJ49+Ar9msOGK6pp0UKs8vjKXFVs0A0hBJY8D+iMSnd8ufa6JjnOUs/
+y6r//Axt6/U64SrMDpCLFyjjnySgtF34W12KNrYocTLWNn9mFVMKXaA1jreVMz4v
+SUueHAmWsD38L2vV7RL9jrwfRSjatMpor68WSbUCgYEAoH16QdkrFU6jvP9rcd62
+UjtzUvg3Lpk/727qiBbLOnqGUMT9dNelC7tHyEkv+EgBtIXi02WVIPXdYgW/nOqQ
+zBtkIdOF+rpVgwfmN6l6uASlp3nJNvLPm7i1kCr4XiXFvCkMDg1RRJul8wfC8+Uz
+wDMtidTTy+HHcN8z+RtAYTI=
+-----END PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIDRDCCAiygAwIBAgIUaDYoj/h5DT4nKKC+aXXMogW23xEwDQYJKoZIhvcNAQEL
+BQAwODEQMA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3Qx
+CzAJBgNVBAYTAkNOMB4XDTI2MDUxMTA3NDU0MFoXDTM2MDUwODA3NDU0MFowPDEU
+MBIGA1UEAwwLVGVzdCBDbGllbnQxFzAVBgNVBAoMDkZsaW5rIENEQyBUZXN0MQsw
+CQYDVQQGEwJDTjCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAJdonYSK
+eHk/TDjwqBp4MMG1PipiemxUNIeQP7UFmBs8Ehr2F9Do8WFSl61PfkrZpJIjjVIm
+kw8nW47i49E11pjIKOfe6zpEqT1cYt6gOreQRujmK7C1H7RAHbwdRPQcTvLa6m5T
+p79waeDglbMHvDMrw7uJLq1qEFSz+g5FEoBaIcxwap3m4FgapcmlMrdKxZKyQBc6
+Kwho3+qhcbybWLYi2r3djduKx943XG7YBASjR3rVn94IeCF40qBsq4z0FCaWe4lx
+xasEXEOE3JSUXq1qKHFAEywmW4W427DTeNWeF0MZwNT8amEOFXN18gglY94piw+d
+n7US1h6qVCK4S80CAwEAAaNCMEAwHQYDVR0OBBYEFGhHtur41a1Ls20y0f2uavbb
+Aey3MB8GA1UdIwQYMBaAFECmXAyBG7NV09/0PoL9rhlI1fpIMA0GCSqGSIb3DQEB
+CwUAA4IBAQAihv58yUnc2P5tXevUB4g8LoC4ZH+nBX9Nl7ZWJX6a7fkA4kH5O/lL
+DVlS0Xs1NZ6MbYoSQTZnpJ1N/Lh/76ens2TxExznmwWy7+JJgpjPivpbbi5wIEwJ
+Vkn8RRlrCobRBSRacA2Jn11ZPEC7EqcNoF7t3SR1hPDT4EJngbggavhhW+F2mPY9
+B5ZROfFVEjSTlfXt6g/FVZ/HL7CC7vc82R5WCYvGkIps6bR+PHUqP4yNXTW6L/Ca
+qeEpJNjJ1SwmpVl++MTtSj1OPOOLzXXzmYHylEDAzNPQlnSIeh7AJsrUfeLwrUFO
+07EbLxeRUYa4VLfWDO/oL1deaMgyyr5M
+-----END CERTIFICATE-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.crt
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.crt
new file mode 100644
index 000000000..75110ed39
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.crt
@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDXjCCAkagAwIBAgIUaDYoj/h5DT4nKKC+aXXMogW23xAwDQYJKoZIhvcNAQEL
+BQAwODEQMA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3Qx
+CzAJBgNVBAYTAkNOMB4XDTI2MDUxMTA3NDUzOVoXDTM2MDUwODA3NDUzOVowOjES
+MBAGA1UEAwwJbG9jYWxob3N0MRcwFQYDVQQKDA5GbGluayBDREMgVGVzdDELMAkG
+A1UEBhMCQ04wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDyDlD250vq
+JSQPEptVBBkvHfUGOiSqcxBi6MP0SKGSxbZ0Qu/g/Ms0Md1zBTar+Ot0IX2X3Txg
+Z8PkrnkRQG44RmdkNuqtd3AoHlMpbeWU/4OUWQ4FTKkrcb3kT1PI1MvQTMulruVG
+GFTqG/wxg1LJyDDS4lPGqH7vU+uxAnmTtS9GSH5P7ncJi5gm+F2XMMiuZDjEezrH
+8X/S58fI1TRqmz5F6FA11Z0YWKG3zxnVWSlt6g4CiwSJVDLbnb+/y74hF8VsjXgg
+S9ZWCGIpBt6KcPhdXjBc2xU9D7zBnV0HdTxDGIGqg5ftleNYfrkQlvppfDkqT9z7
+Cl+JRERXZKJLAgMBAAGjXjBcMBoGA1UdEQQTMBGCCWxvY2FsaG9zdIcEfwAAATAd
+BgNVHQ4EFgQUD3lb1n2pFHOhT9ZVEL0BoZGkxIMwHwYDVR0jBBgwFoAUQKZcDIEb
+s1XT3/Q+gv2uGUjV+kgwDQYJKoZIhvcNAQELBQADggEBAC5sP+zBuDu8ABVjPICZ
+TKJ3ilQ/a8snDt1p7TEQwkVfWWu40zsVcKmQ9CHx8TZ93L/0mZq0R4T6xDsCg/1G
+m61Zkz9eXCA9ZrxnsCB4K8VG6/83eOfk4lyjEdp+ZF+VK8Wwh+xlVeq2zBly9EK1
+fO+wpC38iSQPq0q3u02uCVDUBxplDjhVD6Tgbj1vr/RZfP0aULG2npQjNi5LmtM3
+u2PHFcgRuxA35161vIswRKXpUUVll1L8ZaLfZVcWmwzG/P1qCeDHugSNMblkX/Zq
+MKFiyFVGaHebG9B1qK7Zy2xOBP1fgbuNxGvsog5ebny7u+BV/QiSY8mftk3KenqK
++II=
+-----END CERTIFICATE-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.csr
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.csr
new file mode 100644
index 000000000..fb9e846fe
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.csr
@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE REQUEST-----
+MIICfzCCAWcCAQAwOjESMBAGA1UEAwwJbG9jYWxob3N0MRcwFQYDVQQKDA5GbGlu
+ayBDREMgVGVzdDELMAkGA1UEBhMCQ04wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAw
+ggEKAoIBAQDyDlD250vqJSQPEptVBBkvHfUGOiSqcxBi6MP0SKGSxbZ0Qu/g/Ms0
+Md1zBTar+Ot0IX2X3TxgZ8PkrnkRQG44RmdkNuqtd3AoHlMpbeWU/4OUWQ4FTKkr
+cb3kT1PI1MvQTMulruVGGFTqG/wxg1LJyDDS4lPGqH7vU+uxAnmTtS9GSH5P7ncJ
+i5gm+F2XMMiuZDjEezrH8X/S58fI1TRqmz5F6FA11Z0YWKG3zxnVWSlt6g4CiwSJ
+VDLbnb+/y74hF8VsjXggS9ZWCGIpBt6KcPhdXjBc2xU9D7zBnV0HdTxDGIGqg5ft
+leNYfrkQlvppfDkqT9z7Cl+JRERXZKJLAgMBAAGgADANBgkqhkiG9w0BAQsFAAOC
+AQEAtV7js7BX1ltbaWf6M3O29UOCv9O+L53FHOCdKWW4rHzMEYppB0fuv2Suqh59
+jh5zCMpVljzj5f6Xw2DszmfBC7ovcjsgGzUTGFMZi7WzXagruNq/XXOtgak7igcp
+ft4pqhck6BxQi0vk9K/c3XgWXnKzCPB5aE/lm1Q2zsWE6fSLDTJMV7TtrY6FQzjo
+LAvJVK3Iv2/yzYC2FuTv7IVOUYZ64udY1usprLEVi99bSq29es5lmyX4yL9VV63V
+tYTa1nlaH+3cgubg10G6wmIxAZxeIrYBBVt3rgcxkHhrjCMiO8846juxZGFLuWGO
+5IbLJfLNxD5Ve6/c3hqCifpCgQ==
+-----END CERTIFICATE REQUEST-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.ext
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.ext
new file mode 100644
index 000000000..f9223f594
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.ext
@@ -0,0 +1,2 @@
+[SAN]
+subjectAltName=DNS:localhost,IP:127.0.0.1
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.key
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.key
new file mode 100644
index 000000000..ab596185e
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDyDlD250vqJSQP
+EptVBBkvHfUGOiSqcxBi6MP0SKGSxbZ0Qu/g/Ms0Md1zBTar+Ot0IX2X3TxgZ8Pk
+rnkRQG44RmdkNuqtd3AoHlMpbeWU/4OUWQ4FTKkrcb3kT1PI1MvQTMulruVGGFTq
+G/wxg1LJyDDS4lPGqH7vU+uxAnmTtS9GSH5P7ncJi5gm+F2XMMiuZDjEezrH8X/S
+58fI1TRqmz5F6FA11Z0YWKG3zxnVWSlt6g4CiwSJVDLbnb+/y74hF8VsjXggS9ZW
+CGIpBt6KcPhdXjBc2xU9D7zBnV0HdTxDGIGqg5ftleNYfrkQlvppfDkqT9z7Cl+J
+RERXZKJLAgMBAAECggEABGCzIPFdn/OhrUPjKWj2LTIkOkNv8G+vTyDcr0sRp65J
+1qOsY73M6euxLR3BhcNDdKmeLSfJcUGUglk6khcG1YMpylDsnUwzbAc3CtLUm9rA
+SL0QsgRliGnH0kAVs3KlWh7ojYgrgs9q2GcUKxMTkBDMF+xMTbH7cGRQNza5955F
+pkz4yY8w2rmZaMuQhyoNfrwl8MwIDs5KfbDtvrvELXbNfl8VOVtgYwZmc6Ku1IbZ
+do93MFhRtn5LkxcyJK++XyqRezl8idWi29O+MY41MEqbvktRVO4NgVI5pWITmACo
+7vjLvjdDIX6YIpZKfR3JshsqZP/ipy0iuxPbT0o4eQKBgQD6lgrmnS+l0rX/y1+a
+Cw+VzCCJKVbLNLhpap5cl6SPhU8/PZdQPrUtK4RneGwVyBGtNPrk6qZXu9ghfF1j
+MuGRAYO3rp4DxYZcARfGnEQJoQQVq+UKJFJTmJfvm3pbxPLcKXRovK2PzQp+9LN4
+9Inuiec7XyAYufglrCwC1L7GcwKBgQD3SRgsJ+iwSG9brWZk3IGHYQ0yQGxmzYwO
+k2qxo74buHwPxQJWXx/275MmTzhwRi0NscrMKl69sJnlFY4EVbILSA1YJRYVKy5w
+ntgZ3xuhy8WAyjnjFKJK3fEVPW1FqU96KFzApj+e4DqZNN6cz3L+uKanY4a3AYKC
+Y08zMBNmyQKBgHqrDLcL55FTbwjDNA+iroR0ItFCTVy4eDYSnWdSTJp9defNbM85
+g1OTZAOK7c5T3GNf9hotzDRSnmD2EnlHJSjpBuPaNTw/a6CKYGs3gA7+pZM+TKcE
+VNDbeDUxdVqLNSAVowjKpGQqyaKQg3MIsqT1GR71WC8CirlWaU0RfDqFAoGBAMsW
+Bbsl34z/QC1boE9PZTWIuCzACgb34oGKAkL3xka/4WuWEdBbeUX1BbWdtC1JBPFN
+Rc8dMQe6hPlGFYdr1bg8KXFtq9Ag0Zf+bF37BJeBYafe7WGYbZzs33ThKN/gJ/yc
+L2isYG2bm6R9biH8oWWewoeQTm/31nY9Fytj3Vs5AoGAAnqKrR/dSiEHWCq2qbv2
+dRB54TWF2OP3lGLQPLdgmlxbkXZcSWO0OxfGund4d/z30G2Osvh0lpuTwVrTMn1/
+vryD1+/KoMSGF+nRfyXSkIviYv3aGl5C1DQqXqIRaamiyjv54hZVY7H7bP+17EWu
+ZgqW279pY5Fe/nHl8CN73KY=
+-----END PRIVATE KEY-----
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.pem
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.pem
new file mode 100644
index 000000000..a70f6bca4
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/resources/ssl/server.pem
@@ -0,0 +1,49 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDyDlD250vqJSQP
+EptVBBkvHfUGOiSqcxBi6MP0SKGSxbZ0Qu/g/Ms0Md1zBTar+Ot0IX2X3TxgZ8Pk
+rnkRQG44RmdkNuqtd3AoHlMpbeWU/4OUWQ4FTKkrcb3kT1PI1MvQTMulruVGGFTq
+G/wxg1LJyDDS4lPGqH7vU+uxAnmTtS9GSH5P7ncJi5gm+F2XMMiuZDjEezrH8X/S
+58fI1TRqmz5F6FA11Z0YWKG3zxnVWSlt6g4CiwSJVDLbnb+/y74hF8VsjXggS9ZW
+CGIpBt6KcPhdXjBc2xU9D7zBnV0HdTxDGIGqg5ftleNYfrkQlvppfDkqT9z7Cl+J
+RERXZKJLAgMBAAECggEABGCzIPFdn/OhrUPjKWj2LTIkOkNv8G+vTyDcr0sRp65J
+1qOsY73M6euxLR3BhcNDdKmeLSfJcUGUglk6khcG1YMpylDsnUwzbAc3CtLUm9rA
+SL0QsgRliGnH0kAVs3KlWh7ojYgrgs9q2GcUKxMTkBDMF+xMTbH7cGRQNza5955F
+pkz4yY8w2rmZaMuQhyoNfrwl8MwIDs5KfbDtvrvELXbNfl8VOVtgYwZmc6Ku1IbZ
+do93MFhRtn5LkxcyJK++XyqRezl8idWi29O+MY41MEqbvktRVO4NgVI5pWITmACo
+7vjLvjdDIX6YIpZKfR3JshsqZP/ipy0iuxPbT0o4eQKBgQD6lgrmnS+l0rX/y1+a
+Cw+VzCCJKVbLNLhpap5cl6SPhU8/PZdQPrUtK4RneGwVyBGtNPrk6qZXu9ghfF1j
+MuGRAYO3rp4DxYZcARfGnEQJoQQVq+UKJFJTmJfvm3pbxPLcKXRovK2PzQp+9LN4
+9Inuiec7XyAYufglrCwC1L7GcwKBgQD3SRgsJ+iwSG9brWZk3IGHYQ0yQGxmzYwO
+k2qxo74buHwPxQJWXx/275MmTzhwRi0NscrMKl69sJnlFY4EVbILSA1YJRYVKy5w
+ntgZ3xuhy8WAyjnjFKJK3fEVPW1FqU96KFzApj+e4DqZNN6cz3L+uKanY4a3AYKC
+Y08zMBNmyQKBgHqrDLcL55FTbwjDNA+iroR0ItFCTVy4eDYSnWdSTJp9defNbM85
+g1OTZAOK7c5T3GNf9hotzDRSnmD2EnlHJSjpBuPaNTw/a6CKYGs3gA7+pZM+TKcE
+VNDbeDUxdVqLNSAVowjKpGQqyaKQg3MIsqT1GR71WC8CirlWaU0RfDqFAoGBAMsW
+Bbsl34z/QC1boE9PZTWIuCzACgb34oGKAkL3xka/4WuWEdBbeUX1BbWdtC1JBPFN
+Rc8dMQe6hPlGFYdr1bg8KXFtq9Ag0Zf+bF37BJeBYafe7WGYbZzs33ThKN/gJ/yc
+L2isYG2bm6R9biH8oWWewoeQTm/31nY9Fytj3Vs5AoGAAnqKrR/dSiEHWCq2qbv2
+dRB54TWF2OP3lGLQPLdgmlxbkXZcSWO0OxfGund4d/z30G2Osvh0lpuTwVrTMn1/
+vryD1+/KoMSGF+nRfyXSkIviYv3aGl5C1DQqXqIRaamiyjv54hZVY7H7bP+17EWu
+ZgqW279pY5Fe/nHl8CN73KY=
+-----END PRIVATE KEY-----
+-----BEGIN CERTIFICATE-----
+MIIDXjCCAkagAwIBAgIUaDYoj/h5DT4nKKC+aXXMogW23xAwDQYJKoZIhvcNAQEL
+BQAwODEQMA4GA1UEAwwHVGVzdCBDQTEXMBUGA1UECgwORmxpbmsgQ0RDIFRlc3Qx
+CzAJBgNVBAYTAkNOMB4XDTI2MDUxMTA3NDUzOVoXDTM2MDUwODA3NDUzOVowOjES
+MBAGA1UEAwwJbG9jYWxob3N0MRcwFQYDVQQKDA5GbGluayBDREMgVGVzdDELMAkG
+A1UEBhMCQ04wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDyDlD250vq
+JSQPEptVBBkvHfUGOiSqcxBi6MP0SKGSxbZ0Qu/g/Ms0Md1zBTar+Ot0IX2X3Txg
+Z8PkrnkRQG44RmdkNuqtd3AoHlMpbeWU/4OUWQ4FTKkrcb3kT1PI1MvQTMulruVG
+GFTqG/wxg1LJyDDS4lPGqH7vU+uxAnmTtS9GSH5P7ncJi5gm+F2XMMiuZDjEezrH
+8X/S58fI1TRqmz5F6FA11Z0YWKG3zxnVWSlt6g4CiwSJVDLbnb+/y74hF8VsjXgg
+S9ZWCGIpBt6KcPhdXjBc2xU9D7zBnV0HdTxDGIGqg5ftleNYfrkQlvppfDkqT9z7
+Cl+JRERXZKJLAgMBAAGjXjBcMBoGA1UdEQQTMBGCCWxvY2FsaG9zdIcEfwAAATAd
+BgNVHQ4EFgQUD3lb1n2pFHOhT9ZVEL0BoZGkxIMwHwYDVR0jBBgwFoAUQKZcDIEb
+s1XT3/Q+gv2uGUjV+kgwDQYJKoZIhvcNAQELBQADggEBAC5sP+zBuDu8ABVjPICZ
+TKJ3ilQ/a8snDt1p7TEQwkVfWWu40zsVcKmQ9CHx8TZ93L/0mZq0R4T6xDsCg/1G
+m61Zkz9eXCA9ZrxnsCB4K8VG6/83eOfk4lyjEdp+ZF+VK8Wwh+xlVeq2zBly9EK1
+fO+wpC38iSQPq0q3u02uCVDUBxplDjhVD6Tgbj1vr/RZfP0aULG2npQjNi5LmtM3
+u2PHFcgRuxA35161vIswRKXpUUVll1L8ZaLfZVcWmwzG/P1qCeDHugSNMblkX/Zq
+MKFiyFVGaHebG9B1qK7Zy2xOBP1fgbuNxGvsog5ebny7u+BV/QiSY8mftk3KenqK
++II=
+-----END CERTIFICATE-----
diff --git a/pom.xml b/pom.xml
index 6bc39e7a6..2745728b5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -470,6 +470,12 @@ limitations under the License.
<exclude>
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/Dockerfile
</exclude>
+ <!-- Certificate files -->
+ <exclude>**/ssl/*.crt</exclude>
+ <exclude>**/ssl/*.csr</exclude>
+ <exclude>**/ssl/*.ext</exclude>
+ <exclude>**/ssl/*.key</exclude>
+ <exclude>**/ssl/*.srl</exclude>
</excludes>
</configuration>
</plugin>
