[FLINK-39610] Mongo CDC supports specifying individual SSL context

Author: yuxiqian

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/MongoDBSourceBuilder.java

@@ -285,4 +285,52 @@ public MongoDBSourceBuilder<T> assignUnboundedChunkFirst(boolean assignUnbounded
     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;
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfig.java

@@ -57,6 +57,15 @@ public class MongoDBSourceConfig implements SourceConfig {
     private final boolean isScanNewlyAddedTableEnabled;
     private final boolean assignUnboundedChunkFirst;
 
+    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,
@@ -79,7 +88,15 @@ public class MongoDBSourceConfig implements SourceConfig {
             boolean disableCursorTimeout,
             boolean skipSnapshotBackfill,
             boolean isScanNewlyAddedTableEnabled,
-            boolean assignUnboundedChunkFirst) {
+            boolean assignUnboundedChunkFirst,
+            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;
@@ -103,6 +120,14 @@ public class MongoDBSourceConfig implements SourceConfig {
         this.skipSnapshotBackfill = skipSnapshotBackfill;
         this.isScanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled;
         this.assignUnboundedChunkFirst = assignUnboundedChunkFirst;
+        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() {
@@ -207,6 +232,42 @@ public boolean isAssignUnboundedChunkFirst() {
         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 equals(Object o) {
         if (this == o) {
@@ -234,7 +295,15 @@ public boolean equals(Object o) {
                 && Objects.equals(collectionList, that.collectionList)
                 && Objects.equals(connectionString, that.connectionString)
                 && Objects.equals(skipSnapshotBackfill, that.skipSnapshotBackfill)
-                && Objects.equals(isScanNewlyAddedTableEnabled, that.isScanNewlyAddedTableEnabled);
+                && Objects.equals(isScanNewlyAddedTableEnabled, that.isScanNewlyAddedTableEnabled)
+                && 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
@@ -258,6 +327,14 @@ public int hashCode() {
                 samplesPerChunk,
                 closeIdleReaders,
                 skipSnapshotBackfill,
-                isScanNewlyAddedTableEnabled);
+                isScanNewlyAddedTableEnabled,
+                sslEnabled,
+                sslInvalidHostnameAllowed,
+                sslKeyStore,
+                sslKeyStorePassword,
+                sslKeyStoreType,
+                sslTrustStore,
+                sslTrustStorePassword,
+                sslTrustStoreType);
     }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/config/MongoDBSourceConfigFactory.java

@@ -64,6 +64,17 @@ public class MongoDBSourceConfigFactory implements Factory<MongoDBSourceConfig>
     protected boolean scanNewlyAddedTableEnabled = false;
     protected boolean assignUnboundedChunkFirst = false;
 
+    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(
@@ -280,6 +291,54 @@ public MongoDBSourceConfigFactory assignUnboundedChunkFirst(boolean assignUnboun
         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) {
@@ -306,6 +365,14 @@ public MongoDBSourceConfig create(int subtaskId) {
                 disableCursorTimeout,
                 skipSnapshotBackfill,
                 scanNewlyAddedTableEnabled,
-                assignUnboundedChunkFirst);
+                assignUnboundedChunkFirst,
+                sslEnabled,
+                sslInvalidHostnameAllowed,
+                sslKeyStore,
+                sslKeyStorePassword,
+                sslKeyStoreType,
+                sslTrustStore,
+                sslTrustStorePassword,
+                sslTrustStoreType);
     }
 }

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.");
 }