feat: add support for SSL mode in postgres connector

src/connector/Cargo.toml

@@ -83,8 +83,10 @@ mysql_common = { version = "0.31", default-features = false, features = [
 nexmark = { version = "0.2", features = ["serde"] }
 num-bigint = "0.4"
 opendal = "0.44"
+openssl = "0.10"
 parking_lot = "0.12"
 paste = "1"
+postgres-openssl = "0.5.0"
 prometheus = { version = "0.13", features = ["process"] }
 prost = { version = "0.12", features = ["no-recursion-limit"] }
 prost-reflect = "0.13"

src/connector/src/error.rs

@@ -61,6 +61,8 @@ def_anyhow_newtype! {
     tokio_rustls::rustls::Error => "TLS error",
     rumqttc::v5::ClientError => "MQTT error",
     rumqttc::v5::OptionError => "MQTT error",
+
+    openssl::error::ErrorStack => "OpenSSL error",
 }
 
 pub type ConnectorResult<T, E = ConnectorError> = std::result::Result<T, E>;

src/connector/src/source/cdc/external/mod.rs

@@ -246,7 +246,7 @@ pub struct ExternalTableConfig {
     pub schema: String,
     #[serde(rename = "table.name")]
     pub table: String,
-    #[serde(rename = "ssl.name", default = "Default::default")]
+    #[serde(rename = "ssl.mode", default = "Default::default")]
     pub sslmode: SslMode,
 }
 

src/connector/src/source/cdc/external/postgres.rs

@@ -20,13 +20,14 @@ use futures::stream::BoxStream;
 use futures::{pin_mut, StreamExt};
 use futures_async_stream::try_stream;
 use itertools::Itertools;
+use openssl::ssl::{SslConnector, SslMethod};
+use postgres_openssl::MakeTlsConnector;
 use risingwave_common::catalog::Schema;
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::types::DatumRef;
 use serde_derive::{Deserialize, Serialize};
 use thiserror_ext::AsReport;
 use tokio_postgres::types::PgLsn;
-use tokio_postgres::NoTls;
 
 use crate::error::{ConnectorError, ConnectorResult};
 use crate::parser::postgres_row_to_owned_row;
@@ -135,10 +136,13 @@ impl PostgresExternalTableReader {
             config.host,
             config.port,
             config.database,
-            dbg!(&config.sslmode)
+            config.sslmode
         );
 
-        let (client, connection) = tokio_postgres::connect(&database_url, NoTls).await?;
+        let builder = SslConnector::builder(SslMethod::tls())?;
+        let connector = MakeTlsConnector::new(builder.build());
+
+        let (client, connection) = tokio_postgres::connect(&database_url, connector).await?;
 
         tokio::spawn(async move {
             if let Err(e) = connection.await {