Apache Flink 1.18 Mongo Sink Connector SSL配置问题求助

因为Flink官方MongoSink的MongoWriter仅通过连接URI初始化MongoClient，无法直接通过代码调用applyToSslSettings配置SSL上下文，可通过以下两种方案解决：

方案一：通过连接URI嵌入SSL配置 #

MongoDB连接URI支持直接携带SSL相关参数，将DocumentDb的SSL要求配置到URI中，即可让MongoClients.create()自动加载SSL设置，无需修改代码。

需在URI中添加以下核心参数：

• ssl=true：强制启用SSL连接
• sslInvalidHostNameAllowed=true：DocumentDb的证书主机名通常与连接端点不一致，需允许该情况（若环境要求严格，可将DocumentDb的CA证书导入Flink集群信任库后移除该参数）
• sslCAFile=/path/to/rds-combined-ca-bundle.pem：指定DocumentDb官方CA证书的本地路径（需将证书部署到Flink集群所有TaskManager节点的对应路径）

示例完整URI：

mongodb://<用户名>:<密码>@<DocumentDb端点>:<端口>/<数据库名>?ssl=true&sslInvalidHostNameAllowed=true&sslCAFile=/opt/flink/certs/rds-combined-ca-bundle.pem

注意：需确保Flink集群所有节点都能访问到指定路径的CA证书，可通过集群初始化脚本或Flink分布式缓存提前部署证书文件。

方案二：自定义MongoSink实现 #

若URI参数配置无法满足需求，可修改官方MongoWriter的初始化逻辑，手动配置SSL上下文：

1. 复制官方MongoWriter代码，修改MongoClient初始化部分：

// 初始化带SSL配置的MongoClient
MongoClientSettings settings = MongoClientSettings.builder()
        .applyConnectionString(new ConnectionString(connectionOptions.getUri()))
        .applyToSslSettings(builder -> {
            try {
                // 加载DocumentDb CA证书
                CertificateFactory cf = CertificateFactory.getInstance("X.509");
                File caFile = new File("/path/to/rds-combined-ca-bundle.pem");
                Certificate ca;
                try (InputStream in = new FileInputStream(caFile)) {
                    ca = cf.generateCertificate(in);
                }

                TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
                KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
                ks.load(null, null);
                ks.setCertificateEntry("ca", ca);
                tmf.init(ks);

                builder.enabled(true)
                        .trustManager(tmf)
                        .invalidHostNameAllowed(true); // 根据实际环境调整
            } catch (Exception e) {
                throw new RuntimeException("SSL配置初始化失败", e);
            }
        })
        .build();

this.mongoClient = MongoClients.create(settings);

2. 自定义MongoSink类，使用修改后的MongoWriter，并在Flink作业中替换默认MongoSink为自定义实现。

3. 将自定义Sink代码打包为JAR，上传至Flink集群的lib目录，或提交作业时指定依赖JAR。

验证建议 #

• 优先尝试方案一，无需修改代码，配置成本更低
• 若URI配置无效，再采用方案二的自定义实现
• 确保Flink集群安全组允许访问DocumentDb的默认端口（27017）

内容的提问来源于stack exchange，提问作者Fred